add celery routing queues

Author: giancarloromeo

services/docker-compose.yml

@@ -1193,10 +1193,21 @@ services:
     image: ${DOCKER_REGISTRY:-itisfoundation}/storage:${DOCKER_IMAGE_TAG:-master-github-latest}
     init: true
     hostname: "sto-worker-{{.Node.Hostname}}-{{.Task.Slot}}"
+    environment:
+      <<: *storage_environment
+      STORAGE_WORKER_MODE: "true"
+      CELERY_CONCURRENCY: 100
+    networks: *storage_networks
+
+  sto-worker-low-prio:
+    image: ${DOCKER_REGISTRY:-itisfoundation}/storage:${DOCKER_IMAGE_TAG:-master-github-latest}
+    init: true
+    hostname: "sto-worker-low-prio-{{.Node.Hostname}}-{{.Task.Slot}}"
     environment:
       <<: *storage_environment
       STORAGE_WORKER_MODE: "true"
       CELERY_CONCURRENCY: 1
+      CELERY_QUEUES: "low-prio"
     networks: *storage_networks
 
   rabbit:

services/storage/docker/boot.sh

@@ -54,13 +54,15 @@ if [ "${STORAGE_WORKER_MODE}" = "true" ]; then
       --app=simcore_service_storage.modules.celery.worker_main:app \
       worker --pool=threads \
       --loglevel="${SERVER_LOG_LEVEL}" \
-      --concurrency="${CELERY_CONCURRENCY}"
+      --concurrency="${CELERY_CONCURRENCY}" \
+      --queues="${CELERY_QUEUES:-default}"
   else
     exec celery \
       --app=simcore_service_storage.modules.celery.worker_main:app \
       worker --pool=threads \
       --loglevel="${SERVER_LOG_LEVEL}" \
-      --concurrency="${CELERY_CONCURRENCY}"
+      --concurrency="${CELERY_CONCURRENCY}" \
+      --queues="${CELERY_QUEUES:-default}"
   fi
 else
   if [ "${SC_BOOT_MODE}" = "debug" ]; then

services/storage/src/simcore_service_storage/api/_worker_tasks/tasks.py

@@ -20,7 +20,7 @@ def setup_worker_tasks(app: Celery) -> None:
         logging.INFO,
         msg="Storage setup Worker Tasks",
     ):
-        define_task(app, export_data)
+        define_task(app, export_data, task_queue="low_prio")
         define_task(app, compute_path_size)
         define_task(app, delete_paths)
         define_task(app, complete_upload_file)

services/storage/src/simcore_service_storage/modules/celery/_common.py

@@ -15,6 +15,7 @@ def _celery_configure(celery_settings: CelerySettings) -> dict[str, Any]:
         "result_expires": celery_settings.CELERY_RESULT_EXPIRES,
         "result_extended": True,
         "result_serializer": "json",
+        "task_default_queue": "default",
         "task_send_sent_event": True,
         "task_track_started": True,
         "worker_send_task_events": True,

services/storage/src/simcore_service_storage/modules/celery/_task.py

@@ -82,6 +82,7 @@ def define_task(
     app: Celery,
     fn: Callable[Concatenate[AbortableTask, TaskId, P], Coroutine[Any, Any, R]],
     task_name: str | None = None,
+    task_queue: str | None = None,
 ) -> None: ...
 
 
@@ -90,6 +91,7 @@ def define_task(
     app: Celery,
     fn: Callable[Concatenate[AbortableTask, P], R],
     task_name: str | None = None,
+    task_queue: str | None = None,
 ) -> None: ...
 
 
@@ -100,6 +102,7 @@ def define_task(  # type: ignore[misc]
         | Callable[Concatenate[AbortableTask, P], R]
     ),
     task_name: str | None = None,
+    task_queue: str | None = None,
 ) -> None:
     """Decorator to define a celery task with error handling and abortable support"""
     wrapped_fn: Callable[Concatenate[AbortableTask, P], R]
@@ -114,4 +117,5 @@ def define_task(  # type: ignore[misc]
         name=task_name or fn.__name__,
         bind=True,
         base=AbortableTask,
+        queue=task_queue,
     )(wrapped_fn)