feat: add two kind of workers: worker_high and worker_low

for high and low priority jobs

Makefile

@@ -173,26 +173,26 @@ health:
 i18n-compile:
 	@echo "🥫 Compiling translations …"
 # Note it's important to have --no-deps, to avoid launching a concurrent postgres instance
-	${DOCKER_COMPOSE} run --rm --entrypoint bash --no-deps worker -c "cd i18n && . compile.sh"
+	${DOCKER_COMPOSE} run --rm --entrypoint bash --no-deps worker_high -c "cd i18n && . compile.sh"
 
 unit-tests:
 	@echo "🥫 Running tests …"
 	# run tests in worker to have more memory
 	# also, change project name to run in isolation
-	${DOCKER_COMPOSE_TEST} run --rm worker poetry run pytest --cov-report xml --cov=robotoff tests/unit
+	${DOCKER_COMPOSE_TEST} run --rm worker_high poetry run pytest --cov-report xml --cov=robotoff tests/unit
 
 integration-tests:
 	@echo "🥫 Running integration tests …"
 	# run tests in worker to have more memory
 	# also, change project name to run in isolation
-	${DOCKER_COMPOSE_TEST} run --rm worker poetry run pytest -vv --cov-report xml --cov=robotoff --cov-append tests/integration
+	${DOCKER_COMPOSE_TEST} run --rm worker_high poetry run pytest -vv --cov-report xml --cov=robotoff --cov-append tests/integration
 	( ${DOCKER_COMPOSE_TEST} down -v || true )
 
 # interactive testings
 # usage: make pytest args='test/unit/my-test.py --pdb'
 pytest: guard-args
 	@echo "🥫 Running test: ${args} …"
-	${DOCKER_COMPOSE_TEST} run --rm worker poetry run pytest ${args}
+	${DOCKER_COMPOSE_TEST} run --rm worker_high poetry run pytest ${args}
 
 #------------#
 # Production #

docker-compose.yml

@@ -41,22 +41,33 @@ services:
     <<: *robotoff-base
     environment: *robotoff-base-env
     mem_limit: 2g
-    depends_on:
-      - worker
     ports:
       - "${ROBOTOFF_EXPOSE:-5500}:5500"
     networks:
       - webnet
 
-  worker:
+  worker_high:
     <<: *robotoff-base
     deploy:
       mode: replicated
-      replicas: 8
-    command: poetry run robotoff-cli run-worker
-    environment:
-      <<: *robotoff-base-env
-      REAL_TIME_IMAGE_PREDICTION: 1
+      replicas: 6
+    command: poetry run robotoff-cli run-worker robotoff-high
+    environment: *robotoff-base-env
+    depends_on:
+      - postgres
+    mem_limit: 8g
+    networks:
+      - webnet
+    extra_hosts:
+      - host.docker.internal:host-gateway
+
+  worker_low:
+    <<: *robotoff-base
+    deploy:
+      mode: replicated
+      replicas: 2
+    command: poetry run robotoff-cli run-worker robotoff-low robotoff-high
+    environment: *robotoff-base-env
     depends_on:
       - postgres
     mem_limit: 8g

docker/dev.yml

@@ -50,7 +50,14 @@ services:
         - robotoff.openfoodfacts.localhost
         - api
       webnet:
-  worker:
+  worker_high:
+    <<: *robotoff-dev
+    <<: *networks-productopener-local
+    deploy:
+      mode: replicated
+      # Only 1 replica is easier to deal with for local dev
+      replicas: 1
+  worker_low:
     <<: *robotoff-dev
     <<: *networks-productopener-local
     deploy:

robotoff/cli/main.py

@@ -1,3 +1,4 @@
+import enum
 import pathlib
 import sys
 from pathlib import Path
@@ -19,15 +20,23 @@ def run_scheduler():
     scheduler.run()
 
 
+class WorkerQueue(enum.Enum):
+    robotoff_high = "robotoff-high"
+    robotoff_low = "robotoff-low"
+
+
 @app.command()
 def run_worker(
+    queues: list[WorkerQueue] = typer.Argument(
+        ..., help="Names of the queues to listen to"
+    ),
     burst: bool = typer.Option(
         False, help="Run in burst mode (quit after all work is done)"
-    )
+    ),
 ):
     from robotoff.workers.main import run
 
-    run(burst=burst)
+    run(queues=[x.value for x in queues], burst=burst)
 
 
 @app.command()

robotoff/workers/main.py

@@ -5,7 +5,7 @@
 from robotoff import settings
 from robotoff.models import with_db
 from robotoff.utils import get_logger
-from robotoff.workers.queues import queue_names, redis_conn
+from robotoff.workers.queues import redis_conn
 
 logger = get_logger()
 settings.init_sentry()
@@ -41,11 +41,11 @@ def run_maintenance_tasks(self):
         load_resources(refresh=True)
 
 
-def run(burst: bool = False):
+def run(queues: list[str], burst: bool = False):
     load_resources()
     try:
         with Connection(connection=redis_conn):
-            w = CustomWorker(queues=queue_names)
+            w = CustomWorker(queues=queues)
             w.work(logging_level="INFO", burst=burst)
     except ConnectionError as e:
         print(e)

robotoff/workers/queues.py

@@ -1,3 +1,4 @@
+import enum
 import threading
 import time
 from typing import Callable, Optional
@@ -7,9 +8,14 @@
 
 from robotoff.redis import redis_conn
 
-queue_names = ["robotoff-high", "robotoff-low"]
-high_queue = Queue("robotoff-high", connection=redis_conn)
-low_queue = Queue("robotoff-low", connection=redis_conn)
+
+class AvailableQueue(enum.Enum):
+    robotoff_high = "robotoff-high"
+    robotoff_low = "robotoff-low"
+
+
+high_queue = Queue(AvailableQueue.robotoff_high.value, connection=redis_conn)
+low_queue = Queue(AvailableQueue.robotoff_low.value, connection=redis_conn)
 
 
 def enqueue_in_job(