Graceful Shutdown: Queue wartet auf laufende Jobs + stop_grace_period 5m
This commit is contained in:
parent
d24949740b
commit
2dc504ffea
@ -128,6 +128,15 @@ async def startup():
|
|||||||
from .queue import start_worker, re_enqueue_pending
|
from .queue import start_worker, re_enqueue_pending
|
||||||
await re_enqueue_pending()
|
await re_enqueue_pending()
|
||||||
start_worker()
|
start_worker()
|
||||||
|
|
||||||
|
|
||||||
|
@app.on_event("shutdown")
|
||||||
|
async def shutdown():
|
||||||
|
"""Graceful Shutdown: warte auf laufende Queue-Jobs bevor der Container stirbt."""
|
||||||
|
from .queue import graceful_shutdown
|
||||||
|
await graceful_shutdown(timeout=300)
|
||||||
|
|
||||||
|
|
||||||
# JSON import disabled - all assessments now live in SQLite DB only
|
# JSON import disabled - all assessments now live in SQLite DB only
|
||||||
# Legacy import would overwrite new v5 assessments with old format
|
# Legacy import would overwrite new v5 assessments with old format
|
||||||
# count = await import_json_assessments(settings.data_dir / "assessments")
|
# count = await import_json_assessments(settings.data_dir / "assessments")
|
||||||
|
|||||||
21
app/queue.py
21
app/queue.py
@ -158,6 +158,27 @@ def start_worker() -> list[asyncio.Task]:
|
|||||||
return _worker_tasks
|
return _worker_tasks
|
||||||
|
|
||||||
|
|
||||||
|
async def graceful_shutdown(timeout: int = 300):
|
||||||
|
"""Wait for running jobs to finish before shutdown.
|
||||||
|
|
||||||
|
Called from FastAPI shutdown event. Waits up to `timeout` seconds
|
||||||
|
for the queue to drain and workers to finish their current job.
|
||||||
|
"""
|
||||||
|
pending = _queue.qsize()
|
||||||
|
processing = sum(1 for j in _jobs.values() if j.get("status") == "processing")
|
||||||
|
if pending == 0 and processing == 0:
|
||||||
|
logger.info("Queue empty, shutdown immediately")
|
||||||
|
return
|
||||||
|
|
||||||
|
logger.warning("Graceful shutdown: waiting for %d pending + %d processing jobs (max %ds)",
|
||||||
|
pending, processing, timeout)
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(_queue.join(), timeout=timeout)
|
||||||
|
logger.info("Queue drained, shutdown clean")
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
logger.error("Graceful shutdown timeout after %ds, %d jobs still pending", timeout, _queue.qsize())
|
||||||
|
|
||||||
|
|
||||||
async def re_enqueue_pending():
|
async def re_enqueue_pending():
|
||||||
"""Mark stale queued jobs from previous run."""
|
"""Mark stale queued jobs from previous run."""
|
||||||
import aiosqlite
|
import aiosqlite
|
||||||
|
|||||||
@ -3,6 +3,7 @@ services:
|
|||||||
build: .
|
build: .
|
||||||
container_name: gwoe-antragspruefer
|
container_name: gwoe-antragspruefer
|
||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
|
stop_grace_period: 5m # Queue-Jobs zu Ende laufen lassen vor Kill
|
||||||
environment:
|
environment:
|
||||||
- DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY}
|
- DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY}
|
||||||
- KEYCLOAK_URL=https://sso.toppyr.de
|
- KEYCLOAK_URL=https://sso.toppyr.de
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user