amiirkhl/tabdeal-job-queue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(worker): add configurable debug logging

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-06-21 02:04:19 +03:30
parent 7a2c26a4e6
commit 201196ffb0
10 changed files with 122 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.env.sample
View File

@@ -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

13
README.md
View File

@@ -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

1
backend/.env.sample
View File

@@ -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

3
backend/README.md
View File

@@ -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.

1
backend/config/settings.py
View File

@@ -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()

29
backend/jobs/handlers.py
View File

@@ -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

14
backend/jobs/management/commands/run_job_workers.py
View File

@@ -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."))

46
backend/jobs/services.py
View File

@@ -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

20
backend/jobs/worker.py
View File

@@ -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)

1
docker-compose.yml
View File

@@ -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