add separate rabbitmq queue for api-worker

Author: bisgaard-itis

packages/service-library/src/servicelib/celery/models.py

@@ -28,6 +28,7 @@ class TaskState(StrEnum):
 class TasksQueue(StrEnum):
     CPU_BOUND = "cpu_bound"
     DEFAULT = "default"
+    API_WORKER_QUEUE = "api_worker_queue"
 
 
 class TaskMetadata(BaseModel):

services/api-server/src/simcore_service_api_server/api/routes/functions_routes.py

@@ -21,7 +21,7 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
-from servicelib.celery.models import TaskFilter, TaskMetadata
+from servicelib.celery.models import TaskFilter, TaskMetadata, TasksQueue
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
 from servicelib.long_running_tasks.models import TaskGet
 
@@ -356,6 +356,8 @@ async def run_function(  # noqa: PLR0913
     task_uuid = await task_manager.submit_task(
         TaskMetadata(
             name=task_name,
+            ephemeral=True,
+            queue=TasksQueue.API_WORKER_QUEUE,
         ),
         task_filter=task_filter,
         user_identity=user_identity,

services/api-server/src/simcore_service_api_server/celery/worker_main.py

@@ -16,33 +16,34 @@
 from .worker_tasks.tasks import setup_worker_tasks
 
 
-def app_factory():
-    _settings = ApplicationSettings.create_from_envs()
-
-    setup_loggers(
-        log_format_local_dev_enabled=_settings.API_SERVER_LOG_FORMAT_LOCAL_DEV_ENABLED,
-        logger_filter_mapping=_settings.API_SERVER_LOG_FILTER_MAPPING,
-        tracing_settings=_settings.API_SERVER_TRACING,
-        log_base_level=_settings.log_level,
-        noisy_loggers=None,
-    )
+def _get_settings() -> ApplicationSettings:
+    return ApplicationSettings.create_from_envs()
 
-    assert _settings.API_SERVER_CELERY  # nosec
-    app = create_celery_app(_settings.API_SERVER_CELERY)
 
-    app_server = FastAPIAppServer(app=create_app(_settings))
+_settings = _get_settings()
+
+setup_loggers(
+    log_format_local_dev_enabled=_settings.API_SERVER_LOG_FORMAT_LOCAL_DEV_ENABLED,
+    logger_filter_mapping=_settings.API_SERVER_LOG_FILTER_MAPPING,
+    tracing_settings=_settings.API_SERVER_TRACING,
+    log_base_level=_settings.log_level,
+    noisy_loggers=None,
+)
+
+assert _settings.API_SERVER_CELERY  # nosec
+app = create_celery_app(_settings.API_SERVER_CELERY)
+
+app_server = FastAPIAppServer(app=create_app(_settings))
 
-    def worker_init_wrapper(sender, **_kwargs):
-        assert _settings.API_SERVER_CELERY  # nosec
-        return partial(on_worker_init, app_server, _settings.API_SERVER_CELERY)(
-            sender, **_kwargs
-        )
 
-    worker_init.connect(worker_init_wrapper)
-    worker_shutdown.connect(on_worker_shutdown)
+def worker_init_wrapper(sender, **_kwargs):
+    assert _settings.API_SERVER_CELERY  # nosec
+    return partial(on_worker_init, app_server, _settings.API_SERVER_CELERY)(
+        sender, **_kwargs
+    )
 
-    setup_worker_tasks(app)
-    return app
 
+worker_init.connect(worker_init_wrapper)
+worker_shutdown.connect(on_worker_shutdown)
 
-app = app_factory()
+setup_worker_tasks(app)

services/api-server/tests/unit/celery/conftest.py

@@ -135,6 +135,6 @@ def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
         concurrency=1,
         loglevel="info",
         perform_ping_check=False,
-        queues="default",
+        queues="api_worker_queue",
     ) as worker:
         yield worker

services/api-server/tests/unit/celery/test_functions.py

@@ -25,7 +25,7 @@
     RegisteredProjectFunctionJob,
 )
 from models_library.projects import ProjectID
-from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata
+from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata, TasksQueue
 from servicelib.common_headers import (
     X_SIMCORE_PARENT_NODE_ID,
     X_SIMCORE_PARENT_PROJECT_UUID,
@@ -37,7 +37,7 @@
     get_task_manager,
 )
 from simcore_service_api_server.api.routes.functions_routes import get_function
-from simcore_service_api_server.celery._worker_tasks._functions_tasks import (
+from simcore_service_api_server.celery.worker_tasks.functions_tasks import (
     run_function as run_function_task,
 )
 from simcore_service_api_server.exceptions.backend_errors import BaseBackEndError
@@ -211,7 +211,9 @@ async def test_celery_error_propagation(
     )
     task_manager = get_task_manager(app=app)
     task_uuid = await task_manager.submit_task(
-        task_metadata=TaskMetadata(name="exception_task"),
+        task_metadata=TaskMetadata(
+            name="exception_task", queue=TasksQueue.API_WORKER_QUEUE
+        ),
         task_filter=TaskFilter.model_validate(job_filter.model_dump()),
     )
 

services/docker-compose.yml

@@ -95,6 +95,7 @@ services:
       API_SERVER_WORKER_NAME: "api-worker-{{.Node.Hostname}}-{{.Task.Slot}}-{{.Task.ID}}"
       API_SERVER_WORKER_MODE: "true"
       CELERY_CONCURRENCY: 100
+      CELERY_QUEUES: "api_worker_queue"
     networks: *api_server_networks