continue working

Author: giancarloromeo

services/storage/docker/boot.sh

@@ -57,13 +57,13 @@ if [ "${SC_BOOT_MODE}" = "debug" ]; then
   "
 else
   if [ "${STORAGE_MODE}" = "NORMAL" ]; then
-    exec uvicorn simcore_service_storage.main:the_app \
+    exec uvicorn simcore_service_storage.main: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 \
+      -A simcore_service_storage.main:app \
       worker
       --loglevel="${SERVER_LOG_LEVEL}"
   fi

services/storage/requirements/_base.in

@@ -16,6 +16,7 @@
 
 aioboto3 # s3 storage
 aiofiles # i/o
+asgi_lifespan
 asyncpg # database
 celery[redis]
 httpx

services/storage/requirements/_base.txt

@@ -87,6 +87,8 @@ arrow==1.3.0
     #   -r requirements/../../../packages/models-library/requirements/_base.in
     #   -r requirements/../../../packages/service-library/requirements/../../../packages/models-library/requirements/_base.in
     #   -r requirements/../../../packages/service-library/requirements/_base.in
+asgi-lifespan==2.1.0
+    # via -r requirements/_base.in
 asgiref==3.8.1
     # via opentelemetry-instrumentation-asgi
 asyncpg==0.30.0
@@ -756,7 +758,9 @@ shellingham==1.5.4
 six==1.17.0
     # via python-dateutil
 sniffio==1.3.1
-    # via anyio
+    # via
+    #   anyio
+    #   asgi-lifespan
 sqlalchemy==1.4.54
     # via
     #   -c requirements/../../../packages/aws-library/requirements/../../../packages/common-library/requirements/../../../requirements/constraints.txt

services/storage/requirements/_test.in

@@ -6,7 +6,6 @@
 --constraint _base.txt
 
 
-asgi_lifespan
 asyncpg-stubs
 coverage
 docker

services/storage/requirements/_test.txt

@@ -21,8 +21,6 @@ anyio==4.8.0
     # via
     #   -c requirements/_base.txt
     #   httpx
-asgi-lifespan==2.1.0
-    # via -r requirements/_test.in
 asyncpg==0.30.0
     # via
     #   -c requirements/_base.txt
@@ -337,7 +335,6 @@ sniffio==1.3.1
     # via
     #   -c requirements/_base.txt
     #   anyio
-    #   asgi-lifespan
 sortedcontainers==2.4.0
     # via fakeredis
 sqlalchemy==1.4.54

services/storage/src/simcore_service_storage/core/application.py

@@ -29,7 +29,6 @@
 from ..dsm import setup_dsm
 from ..dsm_cleaner import setup_dsm_cleaner
 from ..exceptions.handlers import set_exception_handlers
-from ..modules.celery.core import setup_celery
 from ..modules.db import setup_db
 from ..modules.long_running_tasks import setup_rest_api_long_running_tasks_for_uploads
 from ..modules.redis import setup as setup_redis
@@ -88,8 +87,6 @@ def create_app(settings: ApplicationSettings) -> FastAPI:
     if settings.STORAGE_CLEANER_INTERVAL_S:
         setup_dsm_cleaner(app)
 
-    setup_celery(app)
-
     if settings.STORAGE_PROFILING:
         app.add_middleware(ProfilerMiddleware)
 

services/storage/src/simcore_service_storage/core/settings.py

@@ -101,6 +101,8 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
         description="is a dictionary that maps specific loggers (such as 'uvicorn.access' or 'gunicorn.access') to a list of _logger message patterns that should be filtered out.",
     )
 
+    STORAGE_MODE: str
+
     @field_validator("LOG_LEVEL", mode="before")
     @classmethod
     def _validate_loglevel(cls, value: str) -> str:

services/storage/src/simcore_service_storage/main.py

@@ -1,23 +1,59 @@
 """Main application to be deployed in for example uvicorn."""
 
+import asyncio
 import logging
 
-from fastapi import FastAPI
+from asgi_lifespan import LifespanManager
+from celery.signals import worker_init, worker_shutdown
 from servicelib.logging_utils import config_all_loggers
 from simcore_service_storage.core.application import create_app
 from simcore_service_storage.core.settings import ApplicationSettings
+from simcore_service_storage.modules.celery.application import create_celery_app
+from simcore_service_storage.modules.celery.tasks import archive
 
-_the_settings = ApplicationSettings.create_from_envs()
+_settings = ApplicationSettings.create_from_envs()
 
 # SEE https://github.com/ITISFoundation/osparc-simcore/issues/3148
-logging.basicConfig(level=_the_settings.log_level)  # NOSONAR
-logging.root.setLevel(_the_settings.log_level)
+logging.basicConfig(level=_settings.log_level)  # NOSONAR
+logging.root.setLevel(_settings.log_level)
 config_all_loggers(
-    log_format_local_dev_enabled=_the_settings.STORAGE_LOG_FORMAT_LOCAL_DEV_ENABLED,
-    logger_filter_mapping=_the_settings.STORAGE_LOG_FILTER_MAPPING,
-    tracing_settings=_the_settings.STORAGE_TRACING,
+    log_format_local_dev_enabled=_settings.STORAGE_LOG_FORMAT_LOCAL_DEV_ENABLED,
+    logger_filter_mapping=_settings.STORAGE_LOG_FILTER_MAPPING,
+    tracing_settings=_settings.STORAGE_TRACING,
 )
 
+_logger = logging.getLogger(__name__)
 
-# SINGLETON FastAPI app
-the_app: FastAPI = create_app(_the_settings)
+fastapi_app = create_app(_settings)
+
+celery_app = create_celery_app(_settings)
+celery_app.task(name="archive")(archive)
+
+
+@worker_init.connect
+def on_worker_init(**_kwargs):
+    loop = asyncio.new_event_loop()
+    # asyncio.set_event_loop(loop)
+
+    async def lifespan():
+        async with LifespanManager(fastapi_app):
+            _logger.error("FastAPI lifespan started")
+            await asyncio.Event().wait()
+
+    fastapi_app.state.lifespan_task = loop.create_task(lifespan())
+    _logger.error("Worker init: FastAPI lifespan task started")
+
+
+@worker_shutdown.connect
+def on_worker_shutdown(**_kwargs):
+    if fastapi_app.state.lifespan_task:
+        fastapi_app.state.lifespan_task.cancel()
+        _logger.info("FastAPI lifespan stopped.")
+
+
+if _settings.STORAGE_MODE == "WORKER":
+    celery_app.conf.fastapi_app = fastapi_app
+    app = celery_app
+else:
+    fastapi_app.state.celery = celery_app
+    app = fastapi_app

services/storage/src/simcore_service_storage/modules/celery/core.py renamed to services/storage/src/simcore_service_storage/modules/celery/application.py

@@ -6,7 +6,7 @@
 from fastapi import FastAPI
 from settings_library.redis import RedisDatabase
 
-from ...core.settings import get_application_settings
+from ...core.settings import ApplicationSettings
 
 _log = logging.getLogger(__name__)
 
@@ -23,29 +23,19 @@ def cancel_task(self, task_id: str):
 
 
 # TODO: move and use new FastAPI lifespan
-def setup_celery(app: FastAPI) -> None:
-    async def on_startup() -> None:
-        settings = get_application_settings(app)
-        assert settings.STORAGE_REDIS
+def create_celery_app(settings: ApplicationSettings) -> Celery:
+    assert settings.STORAGE_REDIS
 
-        redis_dsn = settings.STORAGE_REDIS.build_redis_dsn(
-            RedisDatabase.CELERY_TASKS,
-        )
+    redis_dsn = settings.STORAGE_REDIS.build_redis_dsn(
+        RedisDatabase.CELERY_TASKS,
+    )
 
-        celery_app = Celery(
-            broker=redis_dsn,
-            backend=redis_dsn,
-        )
+    celery_app = Celery(
+        broker=redis_dsn,
+        backend=redis_dsn,
+    )
 
-        task_queue = CeleryTaskQueue(celery_app)
-
-        app.state.task_queue = task_queue
-
-    async def on_shutdown() -> None:
-        _log.warning("Implementing shutdown of celery app")
-
-    app.add_event_handler("startup", on_startup)
-    app.add_event_handler("shutdown", on_shutdown)
+    return celery_app
 
 
 def get_celery_task_queue(app: FastAPI) -> CeleryTaskQueue: