add storage worker

Author: giancarloromeo

services/docker-compose.yml

@@ -1143,6 +1143,45 @@ services:
       S3_ENDPOINT: ${S3_ENDPOINT}
       S3_REGION: ${S3_REGION}
       S3_SECRET_KEY: ${S3_SECRET_KEY}
+      STORAGE_MODE: NORMAL
+      STORAGE_LOGLEVEL: ${STORAGE_LOGLEVEL}
+      STORAGE_MONITORING_ENABLED: 1
+      STORAGE_PROFILING: ${STORAGE_PROFILING}
+      STORAGE_PORT: ${STORAGE_PORT}
+      TRACING_OPENTELEMETRY_COLLECTOR_PORT: ${TRACING_OPENTELEMETRY_COLLECTOR_PORT}
+      TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT: ${TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT}
+    networks:
+      - default
+      - interactive_services_subnet
+      - storage_subnet
+
+  storage-worker:
+    image: ${DOCKER_REGISTRY:-itisfoundation}/storage:${DOCKER_IMAGE_TAG:-latest}
+    init: true
+    hostname: "sto-{{.Node.Hostname}}-{{.Task.Slot}}"
+    environment:
+      BF_API_KEY: ${BF_API_KEY}
+      BF_API_SECRET: ${BF_API_SECRET}
+      DATCORE_ADAPTER_HOST: ${DATCORE_ADAPTER_HOST:-datcore-adapter}
+      LOG_FORMAT_LOCAL_DEV_ENABLED: ${LOG_FORMAT_LOCAL_DEV_ENABLED}
+      LOG_FILTER_MAPPING : ${LOG_FILTER_MAPPING}
+      POSTGRES_DB: ${POSTGRES_DB}
+      POSTGRES_ENDPOINT: ${POSTGRES_ENDPOINT}
+      POSTGRES_HOST: ${POSTGRES_HOST}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+      POSTGRES_PORT: ${POSTGRES_PORT}
+      POSTGRES_USER: ${POSTGRES_USER}
+      REDIS_HOST: ${REDIS_HOST}
+      REDIS_PORT: ${REDIS_PORT}
+      REDIS_SECURE: ${REDIS_SECURE}
+      REDIS_USER: ${REDIS_USER}
+      REDIS_PASSWORD: ${REDIS_PASSWORD}
+      S3_ACCESS_KEY: ${S3_ACCESS_KEY}
+      S3_BUCKET_NAME: ${S3_BUCKET_NAME}
+      S3_ENDPOINT: ${S3_ENDPOINT}
+      S3_REGION: ${S3_REGION}
+      S3_SECRET_KEY: ${S3_SECRET_KEY}
+      STORAGE_MODE: WORKER
       STORAGE_LOGLEVEL: ${STORAGE_LOGLEVEL}
       STORAGE_MONITORING_ENABLED: 1
       STORAGE_PROFILING: ${STORAGE_PROFILING}

services/storage/docker/boot.sh

@@ -56,8 +56,15 @@ if [ "${SC_BOOT_MODE}" = "debug" ]; then
       --log-level \"${SERVER_LOG_LEVEL}\"
   "
 else
-  exec uvicorn simcore_service_storage.main:the_app \
-    --host 0.0.0.0 \
-    --port ${STORAGE_PORT} \
-    --log-level "${SERVER_LOG_LEVEL}"
+  if [ "${STORAGE_MODE}" = "NORMAL" ]; then
+    exec uvicorn simcore_service_storage.main:the_app \
+      --host 0.0.0.0 \
+      --port ${STORAGE_PORT} \
+      --log-level "${SERVER_LOG_LEVEL}"
+  else
+    exec celery \
+      -A simcore_service_storage.modules.celery.worker.main:app \
+      worker
+      --loglevel="${SERVER_LOG_LEVEL}"
+  fi
 fi

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

@@ -0,0 +1,17 @@
+import logging
+
+from celery import Celery
+from settings_library.redis import RedisDatabase
+
+from ...core.settings import ApplicationSettings
+
+_log = logging.getLogger(__name__)
+
+
+def create_celery_app(settings: ApplicationSettings) -> Celery:
+    assert settings.STORAGE_REDIS
+    app = Celery(
+        broker=settings.STORAGE_REDIS.build_redis_dsn(RedisDatabase.CELERY_TASKS),
+        backend=settings.STORAGE_REDIS.build_redis_dsn(RedisDatabase.CELERY_TASKS),
+    )
+    return app

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

@@ -1,13 +1,10 @@
 import logging
-from multiprocessing import Process
-from typing import Callable, cast
+from typing import cast
 
 from celery import Celery
-from celery.apps.worker import Worker
 from celery.result import AsyncResult
 from fastapi import FastAPI
 from settings_library.redis import RedisDatabase
-from simcore_service_storage.modules.celery.tasks import archive
 
 from ...core.settings import get_application_settings
 
@@ -18,11 +15,8 @@ class CeleryTaskQueue:
     def __init__(self, celery_app: Celery):
         self._celery_app = celery_app
 
-    def create_task(self, task_fn: Callable):
-        self._celery_app.task()(task_fn)
-
-    def send_task(self, name: str, **kwargs) -> AsyncResult:
-        return self._celery_app.send_task(name, **kwargs)
+    def send_task(self, name: str, *args, **kwargs) -> AsyncResult:
+        return self._celery_app.send_task(name, args=args, kwargs=kwargs)
 
     def cancel_task(self, task_id: str):
         self._celery_app.control.revoke(task_id)
@@ -34,7 +28,6 @@ async def on_startup() -> None:
         settings = get_application_settings(app)
         assert settings.STORAGE_REDIS
 
-        assert settings.STORAGE_REDIS
         redis_dsn = settings.STORAGE_REDIS.build_redis_dsn(
             RedisDatabase.CELERY_TASKS,
         )
@@ -45,18 +38,9 @@ async def on_startup() -> None:
         )
 
         task_queue = CeleryTaskQueue(celery_app)
-        task_queue.create_task(archive)
 
         app.state.task_queue = task_queue
 
-        # FIXME: Experiment: to start worker in a separate process
-        def worker_process():
-            worker = Worker(app=app.state.celery_app)
-            worker.start()
-
-        worker_proc = Process(target=worker_process)
-        worker_proc.start()
-
     async def on_shutdown() -> None:
         _log.warning("Implementing shutdown of celery app")
 

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

@@ -3,5 +3,6 @@
 _log = logging.getLogger(__name__)
 
 
-def archive(files: list[str]) -> None:
+def archive(files: list[str]) -> str:
     _log.info("Archiving: %s", ", ".join(files))
+    return "".join(files)

services/storage/src/simcore_service_storage/modules/celery/worker/__init__.py

@@ -0,0 +1,12 @@
+from ....core.settings import ApplicationSettings
+from ...celery.tasks import archive
+from ..configurator import create_celery_app
+
+settings = ApplicationSettings.create_from_envs()
+
+app = create_celery_app(settings)
+
+app.task(name="archive")(archive)
+
+
+__all__ = ["app"]