From 201196ffb0205333a5d7d584891f474cfda3532e Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 21 Jun 2026 02:04:19 +0330 Subject: [PATCH] feat(worker): add configurable debug logging --- .env.sample | 1 + README.md | 13 ++++++ backend/.env.sample | 1 + backend/README.md | 3 ++ backend/config/settings.py | 1 + backend/jobs/handlers.py | 29 +++++++++++- .../management/commands/run_job_workers.py | 14 +++++- backend/jobs/services.py | 46 ++++++++++++++++++- backend/jobs/worker.py | 20 ++++++-- docker-compose.yml | 1 + 10 files changed, 122 insertions(+), 7 deletions(-) diff --git a/.env.sample b/.env.sample index 76024d2..77fde46 100644 --- a/.env.sample +++ b/.env.sample @@ -10,3 +10,4 @@ JOB_WORKER_CLEANUP_BATCH_SIZE=100 JOB_WORKER_BACKOFF_BASE_SECONDS=5 JOB_WORKER_BACKOFF_MAX_SECONDS=300 JOB_WORKER_MAX_ATTEMPTS_DEFAULT=3 +JOB_WORKER_LOG_LEVEL=INFO diff --git a/README.md b/README.md index 4476d36..ad02e36 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,19 @@ Create an admin user: docker compose exec backend python manage.py createsuperuser ``` +Worker logs default to `INFO`. To see every poll, claim, lease renewal, progress update, retry, and completion, set: + +```env +JOB_WORKER_LOG_LEVEL=DEBUG +``` + +Then recreate the worker: + +```powershell +docker compose up -d --build worker +docker compose logs -f worker +``` + ## Architecture ```text diff --git a/backend/.env.sample b/backend/.env.sample index 6c8a88b..2227f02 100644 --- a/backend/.env.sample +++ b/backend/.env.sample @@ -15,3 +15,4 @@ JOB_WORKER_CLEANUP_BATCH_SIZE=100 JOB_WORKER_BACKOFF_BASE_SECONDS=5 JOB_WORKER_BACKOFF_MAX_SECONDS=300 JOB_WORKER_MAX_ATTEMPTS_DEFAULT=3 +JOB_WORKER_LOG_LEVEL=INFO diff --git a/backend/README.md b/backend/README.md index c276b36..6ca447f 100644 --- a/backend/README.md +++ b/backend/README.md @@ -52,4 +52,7 @@ JOB_WORKER_CLEANUP_INTERVAL_SECONDS=5 JOB_WORKER_BACKOFF_BASE_SECONDS=5 JOB_WORKER_BACKOFF_MAX_SECONDS=300 JOB_WORKER_MAX_ATTEMPTS_DEFAULT=3 +JOB_WORKER_LOG_LEVEL=INFO ``` + +Use `JOB_WORKER_LOG_LEVEL=DEBUG` to log every worker poll, claim, lease renewal, progress update, retry, failure, and completion in the worker terminal. diff --git a/backend/config/settings.py b/backend/config/settings.py index ba61b7c..72e0a92 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -128,3 +128,4 @@ JOB_WORKER_CLEANUP_BATCH_SIZE = int(os.getenv("JOB_WORKER_CLEANUP_BATCH_SIZE", " JOB_WORKER_BACKOFF_BASE_SECONDS = int(os.getenv("JOB_WORKER_BACKOFF_BASE_SECONDS", "5")) JOB_WORKER_BACKOFF_MAX_SECONDS = int(os.getenv("JOB_WORKER_BACKOFF_MAX_SECONDS", "300")) JOB_WORKER_MAX_ATTEMPTS_DEFAULT = int(os.getenv("JOB_WORKER_MAX_ATTEMPTS_DEFAULT", "3")) +JOB_WORKER_LOG_LEVEL = os.getenv("JOB_WORKER_LOG_LEVEL", "INFO").upper() diff --git a/backend/jobs/handlers.py b/backend/jobs/handlers.py index 09b9bee..f235374 100644 --- a/backend/jobs/handlers.py +++ b/backend/jobs/handlers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import time from dataclasses import dataclass @@ -7,6 +8,8 @@ from django.conf import settings from jobs.services import emit_progress, renew_lease +logger = logging.getLogger(__name__) + @dataclass(frozen=True) class HandlerContext: @@ -15,9 +18,17 @@ class HandlerContext: attempt: int def progress(self, percent: int, message: str = "") -> None: + logger.debug( + "Handler progress job=%s attempt=%s percent=%s message=%s.", + self.job_id, + self.attempt, + percent, + message, + ) emit_progress(self.job_id, worker_id=self.worker_id, attempt=self.attempt, percent=percent, message=message) def renew(self) -> None: + logger.debug("Handler renewing lease job=%s attempt=%s worker=%s.", self.job_id, self.attempt, self.worker_id) renew_lease( self.job_id, worker_id=self.worker_id, @@ -28,38 +39,51 @@ class HandlerContext: def _sleep_with_progress(context: HandlerContext, seconds: int, *, renew: bool) -> None: seconds = max(1, seconds) + logger.debug( + "Handler sleep loop started job=%s attempt=%s seconds=%s renew=%s.", + context.job_id, + context.attempt, + seconds, + renew, + ) for elapsed in range(seconds): time.sleep(1) percent = min(95, int(((elapsed + 1) / seconds) * 100)) if renew: context.renew() context.progress(percent, f"{percent}% complete") + logger.debug("Handler sleep loop finished job=%s attempt=%s.", context.job_id, context.attempt) def handle_success(payload: dict, context: HandlerContext) -> dict: + logger.debug("Running success handler job=%s attempt=%s payload=%s.", context.job_id, context.attempt, payload) sleep_seconds = int(payload.get("sleep_seconds", 1)) _sleep_with_progress(context, sleep_seconds, renew=True) return {"ok": True, "mode": "success"} def handle_fail(payload: dict, context: HandlerContext) -> dict: + logger.debug("Running fail handler job=%s attempt=%s payload=%s.", context.job_id, context.attempt, payload) context.progress(25, "Intentional failure started") raise RuntimeError(str(payload.get("error", "Intentional demo failure"))) def handle_slow(payload: dict, context: HandlerContext) -> dict: + logger.debug("Running slow handler job=%s attempt=%s payload=%s.", context.job_id, context.attempt, payload) sleep_seconds = int(payload.get("sleep_seconds", 8)) _sleep_with_progress(context, sleep_seconds, renew=True) return {"ok": True, "mode": "slow", "slept_seconds": sleep_seconds} def handle_timeout(payload: dict, context: HandlerContext) -> dict: + logger.debug("Running timeout handler job=%s attempt=%s payload=%s.", context.job_id, context.attempt, payload) sleep_seconds = int(payload.get("sleep_seconds", settings.JOB_WORKER_LEASE_SECONDS + 10)) _sleep_with_progress(context, sleep_seconds, renew=False) return {"ok": True, "mode": "timeout", "slept_seconds": sleep_seconds} def handle_flaky(payload: dict, context: HandlerContext) -> dict: + logger.debug("Running flaky handler job=%s attempt=%s payload=%s.", context.job_id, context.attempt, payload) fail_until_attempt = int(payload.get("fail_until_attempt", 2)) context.progress(35, "Flaky job evaluated") if context.attempt <= fail_until_attempt: @@ -81,5 +105,8 @@ def execute_handler(job, *, worker_id: str) -> dict: handler = HANDLERS.get(job.type) if handler is None: raise ValueError(f"Unknown job type: {job.type}") + logger.debug("Dispatching job %s type=%s worker=%s attempt=%s.", job.id, job.type, worker_id, job.attempts) context = HandlerContext(job_id=str(job.id), worker_id=worker_id, attempt=job.attempts) - return handler(job.payload or {}, context) + result = handler(job.payload or {}, context) + logger.debug("Handler finished job %s type=%s result=%s.", job.id, job.type, result) + return result diff --git a/backend/jobs/management/commands/run_job_workers.py b/backend/jobs/management/commands/run_job_workers.py index 4bd337b..d8ebde8 100644 --- a/backend/jobs/management/commands/run_job_workers.py +++ b/backend/jobs/management/commands/run_job_workers.py @@ -1,6 +1,7 @@ import logging import signal +from django.conf import settings from django.core.management.base import BaseCommand from jobs.worker import JobWorkerRunner @@ -12,10 +13,19 @@ class Command(BaseCommand): help = "Run the configured threaded job worker." def handle(self, *args, **options): - logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + log_level = getattr(logging, settings.JOB_WORKER_LOG_LEVEL, logging.INFO) + logging.basicConfig( + level=log_level, + format="%(asctime)s %(levelname)s %(threadName)s %(name)s %(message)s", + force=True, + ) runner = JobWorkerRunner() signal.signal(signal.SIGINT, runner.request_stop) signal.signal(signal.SIGTERM, runner.request_stop) - self.stdout.write(self.style.SUCCESS(f"Starting {runner.thread_count} job worker thread(s).")) + self.stdout.write( + self.style.SUCCESS( + f"Starting {runner.thread_count} job worker thread(s) with log level {settings.JOB_WORKER_LOG_LEVEL}." + ) + ) runner.run_forever() self.stdout.write(self.style.SUCCESS("Job worker stopped.")) diff --git a/backend/jobs/services.py b/backend/jobs/services.py index 1406260..70f68a1 100644 --- a/backend/jobs/services.py +++ b/backend/jobs/services.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from dataclasses import dataclass from datetime import timedelta @@ -10,6 +11,8 @@ from django.utils import timezone from jobs.models import Job, JobEvent +logger = logging.getLogger(__name__) + class JobOwnershipLost(Exception): pass @@ -67,6 +70,7 @@ def create_job( if normalized_key: existing = Job.objects.filter(idempotency_key=normalized_key).first() if existing is not None: + logger.info("Idempotent create returned existing job %s key=%s.", existing.id, normalized_key) return JobCreateResult(existing, created=False) try: @@ -81,7 +85,9 @@ def create_job( except IntegrityError: if not normalized_key: raise - return JobCreateResult(Job.objects.get(idempotency_key=normalized_key), created=False) + existing = Job.objects.get(idempotency_key=normalized_key) + logger.info("Idempotent create raced and returned existing job %s key=%s.", existing.id, normalized_key) + return JobCreateResult(existing, created=False) emit_event( job, @@ -94,6 +100,7 @@ def create_job( "idempotency_key": normalized_key, }, ) + logger.info("Created job %s type=%s priority=%s available_at=%s.", job.id, job.type, job.priority, job.available_at) return JobCreateResult(job, created=True) @@ -114,6 +121,7 @@ def claim_next_job(*, worker_id: str, lease_seconds: int | None = None) -> Job | job = queryset.first() if job is None: + logger.debug("No claimable queued job found for worker=%s at %s.", worker_id, now.isoformat()) return None job.status = Job.Status.RUNNING @@ -130,6 +138,14 @@ def claim_next_job(*, worker_id: str, lease_seconds: int | None = None) -> Job | message="Job claimed", data={"locked_until": job.locked_until.isoformat()}, ) + logger.info( + "Claimed job %s type=%s attempt=%s worker=%s locked_until=%s.", + job.id, + job.type, + job.attempts, + worker_id, + job.locked_until, + ) return job @@ -145,6 +161,7 @@ def renew_lease(job_id, *, worker_id: str, attempt: int, lease_seconds: int | No attempts=attempt, ).update(locked_until=locked_until, updated_at=now) if not updated: + logger.debug("Lease renewal rejected job=%s worker=%s attempt=%s.", job_id, worker_id, attempt) return None job = Job.objects.get(id=job_id) @@ -156,6 +173,7 @@ def renew_lease(job_id, *, worker_id: str, attempt: int, lease_seconds: int | No message="Lease renewed", data={"locked_until": locked_until.isoformat()}, ) + logger.debug("Renewed lease job=%s worker=%s attempt=%s locked_until=%s.", job_id, worker_id, attempt, locked_until) return job @@ -172,6 +190,7 @@ def emit_progress(job_id, *, worker_id: str, attempt: int, percent: int, message .first() ) if job is None: + logger.debug("Progress rejected job=%s worker=%s attempt=%s percent=%s.", job_id, worker_id, attempt, percent) return False emit_event( @@ -182,6 +201,7 @@ def emit_progress(job_id, *, worker_id: str, attempt: int, percent: int, message message=message or f"{percent}% complete", data={"percent": percent}, ) + logger.debug("Recorded progress job=%s worker=%s attempt=%s percent=%s.", job_id, worker_id, attempt, percent) return True @@ -202,6 +222,7 @@ def complete_job(job_id, *, worker_id: str, attempt: int, result: dict | None = updated_at=now, ) if not updated: + logger.debug("Completion rejected job=%s worker=%s attempt=%s.", job_id, worker_id, attempt) return None job = Job.objects.get(id=job_id) @@ -213,6 +234,7 @@ def complete_job(job_id, *, worker_id: str, attempt: int, result: dict | None = message="Job succeeded", data={"result": job.result}, ) + logger.info("Marked job %s succeeded worker=%s attempt=%s.", job.id, worker_id, attempt) return job @@ -225,6 +247,7 @@ def fail_job(job_id, *, worker_id: str, attempt: int, error: str) -> Job | None: .first() ) if job is None: + logger.debug("Failure rejected job=%s worker=%s attempt=%s error=%s.", job_id, worker_id, attempt, error) return None if job.attempts < job.max_attempts: @@ -243,6 +266,15 @@ def fail_job(job_id, *, worker_id: str, attempt: int, error: str) -> Job | None: message=error, data={"delay_seconds": int(delay.total_seconds()), "available_at": job.available_at.isoformat()}, ) + logger.info( + "Scheduled retry job=%s attempt=%s/%s worker=%s delay_seconds=%s error=%s.", + job.id, + job.attempts, + job.max_attempts, + worker_id, + int(delay.total_seconds()), + error, + ) return job job.status = Job.Status.FAILED @@ -252,6 +284,7 @@ def fail_job(job_id, *, worker_id: str, attempt: int, error: str) -> Job | None: job.finished_at = now job.save(update_fields=["status", "locked_by", "locked_until", "last_error", "finished_at", "updated_at"]) emit_event(job, JobEvent.Type.FAILED, attempt=attempt, worker_id=worker_id, message=error, data={"error": error}) + logger.info("Marked job %s failed after attempt=%s worker=%s error=%s.", job.id, attempt, worker_id, error) return job @@ -266,6 +299,7 @@ def cleanup_expired_jobs(*, batch_size: int | None = None) -> int: queryset = queryset.select_for_update() expired_jobs = list(queryset[:batch_size]) + logger.debug("Found %s expired running job(s) for cleanup.", len(expired_jobs)) for job in expired_jobs: worker_id = job.locked_by attempt = job.attempts @@ -285,6 +319,14 @@ def cleanup_expired_jobs(*, batch_size: int | None = None) -> int: message="Worker lease expired; job requeued", data={"delay_seconds": int(delay.total_seconds()), "available_at": job.available_at.isoformat()}, ) + logger.info( + "Requeued expired job=%s attempt=%s/%s previous_worker=%s delay_seconds=%s.", + job.id, + attempt, + job.max_attempts, + worker_id, + int(delay.total_seconds()), + ) continue job.status = Job.Status.FAILED @@ -301,6 +343,7 @@ def cleanup_expired_jobs(*, batch_size: int | None = None) -> int: message="Worker lease expired; attempts exhausted", data={"error": "Worker lease expired"}, ) + logger.info("Failed expired job=%s attempt=%s previous_worker=%s.", job.id, attempt, worker_id) return len(expired_jobs) @@ -333,6 +376,7 @@ def retry_failed_job(job_id) -> Job: ] ) emit_event(job, JobEvent.Type.MANUAL_RETRY, message="Failed job manually retried") + logger.info("Manually retried failed job %s.", job.id) return job diff --git a/backend/jobs/worker.py b/backend/jobs/worker.py index 1450416..700b86d 100644 --- a/backend/jobs/worker.py +++ b/backend/jobs/worker.py @@ -27,12 +27,14 @@ class JobWorkerRunner: self.stop_event.set() def start(self): + logger.info("Starting job worker runner with %s thread(s).", self.thread_count) for index in range(self.thread_count): thread = threading.Thread(target=self._run_thread, args=(index,), name=f"job-worker-{index + 1}") thread.start() self.threads.append(thread) def wait(self): + logger.debug("Waiting for worker threads to stop.") while any(thread.is_alive() for thread in self.threads): for thread in self.threads: thread.join(timeout=0.25) @@ -54,12 +56,17 @@ class JobWorkerRunner: now = time.monotonic() if now - last_cleanup >= settings.JOB_WORKER_CLEANUP_INTERVAL_SECONDS: try: - cleanup_expired_jobs(batch_size=settings.JOB_WORKER_CLEANUP_BATCH_SIZE) + cleaned = cleanup_expired_jobs(batch_size=settings.JOB_WORKER_CLEANUP_BATCH_SIZE) + if cleaned: + logger.info("Cleaned up %s expired job lease(s).", cleaned) + else: + logger.debug("Expired job cleanup found no work.") except Exception: logger.exception("Expired job cleanup failed.") last_cleanup = now try: + logger.debug("Polling for the next queued job.") job = claim_next_job(worker_id=worker_id, lease_seconds=settings.JOB_WORKER_LEASE_SECONDS) except Exception: logger.exception("Job claim failed.") @@ -67,6 +74,7 @@ class JobWorkerRunner: continue if job is None: + logger.debug("No eligible job found; sleeping %.3f second(s).", poll_seconds) self.stop_event.wait(poll_seconds) continue @@ -77,14 +85,20 @@ class JobWorkerRunner: def _execute_claimed_job(self, job, *, worker_id: str): attempt = job.attempts + logger.info("Executing job %s type=%s attempt=%s worker=%s.", job.id, job.type, attempt, worker_id) try: result = execute_handler(job, worker_id=worker_id) except Exception as exc: + logger.info("Job %s attempt=%s raised error: %s", job.id, attempt, exc) updated_job = fail_job(job.id, worker_id=worker_id, attempt=attempt, error=str(exc)) if updated_job is None: - logger.info("Worker lost ownership before failing job %s attempt %s.", job.id, attempt) + logger.warning("Worker lost ownership before failing job %s attempt %s.", job.id, attempt) + else: + logger.debug("Failure handling persisted job %s with status=%s.", updated_job.id, updated_job.status) return updated_job = complete_job(job.id, worker_id=worker_id, attempt=attempt, result=result) if updated_job is None: - logger.info("Worker lost ownership before completing job %s attempt %s.", job.id, attempt) + logger.warning("Worker lost ownership before completing job %s attempt %s.", job.id, attempt) + else: + logger.info("Completed job %s attempt=%s worker=%s.", updated_job.id, attempt, worker_id) diff --git a/docker-compose.yml b/docker-compose.yml index 7ad0a56..adbdc5c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -67,6 +67,7 @@ services: JOB_WORKER_BACKOFF_BASE_SECONDS: ${JOB_WORKER_BACKOFF_BASE_SECONDS:-5} JOB_WORKER_BACKOFF_MAX_SECONDS: ${JOB_WORKER_BACKOFF_MAX_SECONDS:-300} JOB_WORKER_MAX_ATTEMPTS_DEFAULT: ${JOB_WORKER_MAX_ATTEMPTS_DEFAULT:-3} + JOB_WORKER_LOG_LEVEL: ${JOB_WORKER_LOG_LEVEL:-INFO} depends_on: backend: condition: service_healthy