Upgrade configuration and Docker stack to v2.0.0

2026-06-22 14:49:56 -04:00
parent 2f2a5e9b4a
commit 0d2ecbd218
19 changed files with 779 additions and 495 deletions
@@ -1,5 +1,8 @@
 """
 Worker ARQ — traitement asynchrone des images via Redis.
+
+Utilise le parser d'URL Redis centralisé de app.workers.redis_client.
+Inclut la Dead Letter Queue pour les jobs échoués après N tentatives.
 """
 import logging
 import asyncio
@@ -12,6 +15,7 @@ from app.config import settings
 from app.database import AsyncSessionLocal
 from app.models.image import Image, ProcessingStatus
 from app.services.pipeline import process_image_pipeline
+from app.workers.redis_client import parse_redis_url
 from sqlalchemy import select

 # Préfixes ARQ
@@ -21,46 +25,83 @@ DEFAULT_QUEUE_NAME = "arq:queue"

 logger = logging.getLogger(__name__)

+
 async def process_image_task(ctx: dict, image_id: int, client_id: str) -> str:
-    """Tâche ARQ : traite une image."""
+    """Tâche ARQ : traite une image. Pousse en DLQ après échecs répétés."""
    job_try = ctx.get("job_try", 1)
    redis = ctx.get("redis")
+    max_retries = getattr(settings, "PIPELINE_MAX_RETRIES", 3)

-    logger.info(f"--- JOB DÉMARRÉ : image_id={image_id} ---")
+    logger.info(
+        "arq.job.started",
+        extra={"image_id": image_id, "client_id": client_id, "attempt": job_try},
+    )

    async with AsyncSessionLocal() as db:
        try:
+            result = await db.execute(select(Image).where(Image.id == image_id))
+            image = result.scalar_one_or_none()
+            if image is None:
+                logger.warning("arq.job.image_not_found", extra={"image_id": image_id})
+                return "SKIPPED: image not found"
+            if image.processing_status == ProcessingStatus.DONE:
+                logger.info("arq.job.already_done", extra={"image_id": image_id})
+                return "SKIPPED: already done"
+
            await process_image_pipeline(image_id, db, redis=redis)
-            logger.info(f"--- JOB TERMINÉ : image_id={image_id} ---")
-            return f"OK"
+            logger.info("arq.job.completed", extra={"image_id": image_id})
+            return "OK"
+
        except Exception as e:
-            logger.error(f"--- JOB ÉCHOUÉ : {str(e)} ---", exc_info=True)
-            raise
+            logger.error(
+                "arq.job.failed",
+                extra={"image_id": image_id, "error": str(e), "attempt": job_try},
+                exc_info=True,
+            )
+
+            # Si c'est la dernière tentative, pousser en DLQ
+            if job_try >= max_retries:
+                from app.workers.dead_letter import push_dead_job
+                await push_dead_job(
+                    redis=redis,
+                    job_name="process_image_task",
+                    args=[image_id, client_id],
+                    kwargs={},
+                    error=str(e),
+                    attempts=job_try,
+                )
+                # Marquer l'image en erreur permanente
+                async with AsyncSessionLocal() as db2:
+                    result2 = await db2.execute(select(Image).where(Image.id == image_id))
+                    img = result2.scalar_one_or_none()
+                    if img:
+                        img.processing_status = ProcessingStatus.ERROR
+                        img.processing_error = f"[DLQ] Échec après {job_try} tentatives: {str(e)[:200]}"
+                        await db2.commit()
+                return f"DLQ: failed after {job_try} attempts"
+
+            raise  # Re-raise pour que ARQ retente
+

 async def on_startup(ctx: dict) -> None:
-    logger.info("Worker started and listening on %s", WorkerSettings.queue_name)
+    logger.info("worker.started", extra={"queue": WorkerSettings.queue_name})
+
+
+async def on_shutdown(ctx: dict) -> None:
+    logger.info("worker.shutdown")

-def _parse_redis_settings() -> RedisSettings:
-    url = settings.REDIS_URL
-    if url.startswith("redis://"): url = url[8:]
-    elif url.startswith("rediss://"): url = url[9:]
-    password, host, port, database = None, "localhost", 6379, 0
-    if "@" in url:
-        auth, url = url.rsplit("@", 1)
-        password = auth.split(":", 1)[1] if ":" in auth else auth
-    if "/" in url:
-        url, db_str = url.split("/", 1)
-        if db_str: database = int(db_str)
-    if ":" in url:
-        host, port_str = url.rsplit(":", 1)
-        port = int(port_str)
-    else: host = url
-    return RedisSettings(host=host, port=port, password=password, database=database)

 class WorkerSettings:
    functions = [func(process_image_task, name="process_image_task")]
-    redis_settings = _parse_redis_settings()
+    redis_settings = RedisSettings(
+        host=parse_redis_url()[0],
+        port=parse_redis_url()[1],
+        password=parse_redis_url()[2],
+        database=parse_redis_url()[3],
+    )
    queue_name = DEFAULT_QUEUE_NAME
    on_startup = on_startup
+    on_shutdown = on_shutdown
    max_jobs = 10
-    job_timeout = 300
+    job_timeout = settings.PIPELINE_TIMEOUT
+    health_check_interval = 30
@@ -1,6 +1,9 @@
 """
 Client Redis partagé — pool de connexions async pour ARQ et Pub/Sub.
+Inclut le parsing d'URL Redis centralisé pour éviter la duplication.
 """
+from typing import Tuple, Optional
+
 from redis.asyncio import ConnectionPool, Redis

 from app.config import settings
@@ -8,10 +11,63 @@ from app.config import settings
 _pool: ConnectionPool | None = None


+def parse_redis_url(url: str | None = None) -> Tuple[str, int, Optional[str], int]:
+    """
+    Parse une URL Redis en (host, port, password, database).
+
+    Format accepté : redis://[user:password@]host[:port][/db]
+    Retourne ('localhost', 6379, None, 0) si l'URL est vide.
+    """
+    if url is None:
+        url = settings.REDIS_URL
+
+    if not url:
+        return "localhost", 6379, None, 0
+
+    if url.startswith("redis://"):
+        url = url[8:]
+    if url.startswith("rediss://"):
+        url = url[9:]
+
+    password: Optional[str] = None
+    if "@" in url:
+        auth_part, url = url.rsplit("@", 1)
+        if ":" in auth_part:
+            password = auth_part.split(":", 1)[1]
+        else:
+            password = auth_part
+
+    database = 0
+    if "/" in url:
+        host_port, db_str = url.split("/", 1)
+        if db_str:
+            try:
+                database = int(db_str)
+            except ValueError:
+                pass
+    else:
+        host_port = url
+
+    host = "localhost"
+    port = 6379
+    if ":" in host_port:
+        host, port_str = host_port.rsplit(":", 1)
+        if port_str:
+            try:
+                port = int(port_str)
+            except ValueError:
+                port = 6379
+    else:
+        host = host_port or "localhost"
+
+    return host, port, password, database
+
+
 async def get_redis_pool() -> Redis:
    """Retourne un client Redis avec pool de connexions partagé."""
    global _pool
    if _pool is None:
+        host, port, password, database = parse_redis_url()
        _pool = ConnectionPool.from_url(
            settings.REDIS_URL,
            max_connections=20,