Merge branch 'master' of github.com:ITISFoundation/osparc-simcore int…

…o service-startup-progress

packages/simcore-sdk/tests/integration/test_node_data_data_manager.py

@@ -29,6 +29,7 @@
     "migration",
     "postgres",
     "storage",
+    "redis",
 ]
 
 pytest_simcore_ops_services_selection = [

packages/simcore-sdk/tests/integration/test_node_ports_common_filemanager.py

@@ -33,6 +33,7 @@
     "migration",
     "postgres",
     "storage",
+    "redis",
 ]
 
 pytest_simcore_ops_services_selection = ["minio", "adminer"]

packages/simcore-sdk/tests/integration/test_node_ports_common_r_clone.py

@@ -27,6 +27,7 @@
     "migration",
     "postgres",
     "storage",
+    "redis",
 ]
 
 pytest_simcore_ops_services_selection = [

packages/simcore-sdk/tests/integration/test_node_ports_v2_nodeports2.py

@@ -41,6 +41,7 @@
     "migration",
     "postgres",
     "storage",
+    "redis",
 ]
 
 pytest_simcore_ops_services_selection = [

services/director-v2/tests/integration/01/test_computation_api.py

@@ -45,6 +45,7 @@
     "postgres",
     "rabbit",
     "storage",
+    "redis",
 ]
 pytest_simcore_ops_services_selection = ["minio", "adminer"]
 

services/director-v2/tests/integration/02/test_dynamic_services_routes.py

@@ -57,6 +57,7 @@
     "rabbit",
     "redis",
     "storage",
+    "redis",
 ]
 pytest_simcore_ops_services_selection = [
     "adminer",

services/docker-compose.yml

@@ -957,6 +957,8 @@ services:
       - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
       - POSTGRES_PORT=${POSTGRES_PORT}
       - POSTGRES_USER=${POSTGRES_USER}
+      - REDIS_HOST=${REDIS_HOST}
+      - REDIS_PORT=${REDIS_PORT}
       - S3_ACCESS_KEY=${S3_ACCESS_KEY}
       - S3_BUCKET_NAME=${S3_BUCKET_NAME}
       - S3_ENDPOINT=${S3_ENDPOINT}

services/dynamic-sidecar/tests/integration/test_modules_long_running_tasks.py

@@ -45,6 +45,7 @@
     "migration",
     "postgres",
     "storage",
+    "redis",
 ]
 
 pytest_simcore_ops_services_selection = [

services/storage/requirements/_test.in

@@ -10,6 +10,7 @@ aioresponses
 coverage
 docker
 faker
+fakeredis[lua]
 moto[server]
 pandas
 pytest

services/storage/requirements/_test.txt

@@ -9,7 +9,9 @@ aiosignal==1.3.1
 antlr4-python3-runtime==4.13.1
     # via moto
 async-timeout==4.0.3
-    # via aiohttp
+    # via
+    #   aiohttp
+    #   redis
 attrs==23.2.0
     # via
     #   aiohttp
@@ -56,6 +58,7 @@ docker==7.0.0
 exceptiongroup==1.2.0
     # via pytest
 faker==24.4.0
+fakeredis==2.21.3
 flask==3.0.2
     # via
     #   flask-cors
@@ -118,6 +121,8 @@ junit-xml==1.9
     # via cfn-lint
 lazy-object-proxy==1.10.0
     # via openapi-spec-validator
+lupa==2.1
+    # via fakeredis
 markupsafe==2.1.5
     # via
     #   jinja2
@@ -201,6 +206,8 @@ pyyaml==6.0.1
     #   jsonschema-path
     #   moto
     #   responses
+redis==5.0.3
+    # via fakeredis
 referencing==0.29.3
     # via
     #   jsonschema
@@ -235,6 +242,8 @@ six==1.16.0
     #   python-dateutil
     #   rfc3339-validator
     #   simcore-service-storage-sdk
+sortedcontainers==2.4.0
+    # via fakeredis
 sqlalchemy==1.4.52
 sqlalchemy2-stubs==0.0.2a38
     # via sqlalchemy

services/storage/src/simcore_service_storage/application.py

@@ -17,6 +17,7 @@
 from .dsm import setup_dsm
 from .dsm_cleaner import setup_dsm_cleaner
 from .long_running_tasks import setup_long_running_tasks
+from .redis import setup_redis
 from .rest import setup_rest
 from .s3 import setup_s3
 from .settings import Settings
@@ -63,6 +64,7 @@ def create(settings: Settings) -> web.Application:
 
     setup_long_running_tasks(app)
     setup_rest(app)
+    setup_redis(app)
 
     if settings.STORAGE_POSTGRES and settings.STORAGE_S3:
         setup_dsm(app)  # core subsystem. Needs s3 and db setups done

services/storage/src/simcore_service_storage/dsm_cleaner.py

@@ -20,54 +20,60 @@
 
 import asyncio
 import logging
-import os
-import socket
-from contextlib import suppress
+from datetime import timedelta
 from typing import cast
 
 from aiohttp import web
+from servicelib.background_task import stop_periodic_task
+from servicelib.logging_utils import log_catch, log_context
+from servicelib.redis_utils import start_exclusive_periodic_task
 
 from .constants import APP_CONFIG_KEY, APP_DSM_KEY
 from .dsm_factory import DataManagerProvider
+from .redis import get_redis_client
 from .settings import Settings
 from .simcore_s3_dsm import SimcoreS3DataManager
 
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
+
+_TASK_NAME_PERIODICALY_CLEAN_DSM = "periodic_cleanup_of_dsm"
 
 
 async def dsm_cleaner_task(app: web.Application) -> None:
-    logger.info("starting dsm cleaner task...")
-    cfg: Settings = app[APP_CONFIG_KEY]
+    _logger.info("starting dsm cleaner task...")
     dsm: DataManagerProvider = app[APP_DSM_KEY]
     simcore_s3_dsm: SimcoreS3DataManager = cast(
         SimcoreS3DataManager, dsm.get(SimcoreS3DataManager.get_location_id())
     )
-    assert cfg.STORAGE_CLEANER_INTERVAL_S  # nosec
-    while await asyncio.sleep(cfg.STORAGE_CLEANER_INTERVAL_S, result=True):
-        try:
-            await simcore_s3_dsm.clean_expired_uploads()
+    try:
+        await simcore_s3_dsm.clean_expired_uploads()
 
-        except asyncio.CancelledError:  # noqa: PERF203
-            logger.info("cancelled dsm cleaner task")
-            raise
-        except Exception:  # pylint: disable=broad-except
-            logger.exception("Unhandled error in dsm cleaner task, restarting task...")
+    except asyncio.CancelledError:  # noqa: PERF203
+        _logger.info("cancelled dsm cleaner task")
+        raise
+    except Exception:  # pylint: disable=broad-except
+        _logger.exception("Unhandled error in dsm cleaner task, restarting task...")
 
 
 def setup_dsm_cleaner(app: web.Application):
     async def _setup(app: web.Application):
-        task = asyncio.create_task(
-            dsm_cleaner_task(app),
-            name=f"dsm_cleaner_task_{socket.gethostname()}_{os.getpid()}",
-        )
-        logger.info("%s created", f"{task=}")
-
-        yield
-
-        logger.debug("stopping %s...", f"{task=}")
-        task.cancel()
-        with suppress(asyncio.CancelledError):
-            await task
-        logger.info("%s stopped.", f"{task=}")
+        with log_context(_logger, logging.INFO, msg="setup dsm cleaner"), log_catch(
+            _logger, reraise=False
+        ):
+            cfg: Settings = app[APP_CONFIG_KEY]
+            assert cfg.STORAGE_CLEANER_INTERVAL_S  # nosec
+
+            storage_background_task = start_exclusive_periodic_task(
+                get_redis_client(app),
+                dsm_cleaner_task,
+                task_period=timedelta(seconds=cfg.STORAGE_CLEANER_INTERVAL_S),
+                retry_after=timedelta(minutes=5),
+                task_name=_TASK_NAME_PERIODICALY_CLEAN_DSM,
+                app=app,
+            )
+
+            yield
+
+            await stop_periodic_task(storage_background_task)
 
     app.cleanup_ctx.append(_setup)

services/storage/src/simcore_service_storage/redis.py

@@ -0,0 +1,35 @@
+import logging
+from typing import cast
+
+from aiohttp import web
+from servicelib.redis import RedisClientSDK
+from settings_library.redis import RedisDatabase, RedisSettings
+
+from .constants import APP_CONFIG_KEY
+from .settings import Settings
+
+_logger = logging.getLogger(__name__)
+
+_APP_REDIS_KEY = "APP_REDIS_KEY"
+
+
+def setup_redis(app: web.Application):
+    async def _setup(app: web.Application):
+        app[_APP_REDIS_KEY] = None
+        settings: Settings = app[APP_CONFIG_KEY]
+        assert settings.STORAGE_REDIS  # nosec
+        redis_settings: RedisSettings = settings.STORAGE_REDIS
+        redis_locks_dsn = redis_settings.build_redis_dsn(RedisDatabase.LOCKS)
+        app[_APP_REDIS_KEY] = client = RedisClientSDK(redis_locks_dsn)
+        await client.setup()
+
+        yield
+
+        if client:
+            await client.shutdown()
+
+    app.cleanup_ctx.append(_setup)
+
+
+def get_redis_client(app: web.Application) -> RedisClientSDK:
+    return cast(RedisClientSDK, app[_APP_REDIS_KEY])

services/storage/src/simcore_service_storage/settings.py

@@ -2,6 +2,7 @@
 from settings_library.base import BaseCustomSettings
 from settings_library.basic_types import LogLevel, PortInt
 from settings_library.postgres import PostgresSettings
+from settings_library.redis import RedisSettings
 from settings_library.s3 import S3Settings
 from settings_library.tracing import TracingSettings
 from settings_library.utils_logging import MixinLoggingSettings
@@ -33,6 +34,8 @@ class Settings(BaseCustomSettings, MixinLoggingSettings):
 
     STORAGE_POSTGRES: PostgresSettings | None = Field(auto_default_from_env=True)
 
+    STORAGE_REDIS: RedisSettings | None = Field(auto_default_from_env=True)
+
     STORAGE_S3: S3Settings | None = Field(auto_default_from_env=True)
 
     STORAGE_TRACING: TracingSettings | None = Field(auto_default_from_env=True)

services/storage/tests/conftest.py

@@ -22,6 +22,7 @@
 from aiohttp.test_utils import TestClient
 from aiopg.sa import Engine
 from faker import Faker
+from fakeredis.aioredis import FakeRedis
 from models_library.api_schemas_storage import (
     FileMetaDataGet,
     FileUploadCompleteFutureResponse,
@@ -39,6 +40,7 @@
 from models_library.users import UserID
 from models_library.utils.fastapi_encoders import jsonable_encoder
 from pydantic import ByteSize, parse_obj_as
+from pytest_mock import MockerFixture
 from pytest_simcore.helpers.utils_assert import assert_status
 from servicelib.aiohttp import status
 from simcore_postgres_database.storage_models import file_meta_data, projects, users
@@ -196,12 +198,19 @@ def app_settings(mock_config) -> Settings:
     return test_app_settings
 
 
+@pytest.fixture
+async def mocked_redis_server(mocker: MockerFixture) -> None:
+    mock_redis = FakeRedis()
+    mocker.patch("redis.asyncio.from_url", return_value=mock_redis)
+
+
 @pytest.fixture
 def client(
     event_loop: asyncio.AbstractEventLoop,
     aiohttp_client: Callable,
     unused_tcp_port_factory: Callable[..., int],
     app_settings: Settings,
+    mocked_redis_server,
 ) -> TestClient:
     app = create(app_settings)
     return event_loop.run_until_complete(

services/storage/tests/unit/test_dsm_cleaner.py

@@ -8,6 +8,7 @@
 import pytest
 from aiohttp.test_utils import TestClient
 from pytest_mock import MockerFixture
+from simcore_service_storage.dsm_cleaner import _TASK_NAME_PERIODICALY_CLEAN_DSM
 
 pytest_simcore_core_services_selection = ["postgres"]
 pytest_simcore_ops_services_selection = ["adminer"]
@@ -35,12 +36,22 @@ def short_dsm_cleaner_interval(monkeypatch: pytest.MonkeyPatch) -> int:
 
 async def test_setup_dsm_cleaner(client: TestClient):
     all_tasks = asyncio.all_tasks()
-    assert any(t.get_name().startswith("dsm_cleaner_task") for t in all_tasks)
+    assert any(
+        t.get_name().startswith(
+            f"exclusive_task_starter_{_TASK_NAME_PERIODICALY_CLEAN_DSM}"
+        )
+        for t in all_tasks
+    )
 
 
 async def test_disable_dsm_cleaner(disable_dsm_cleaner, client: TestClient):
     all_tasks = asyncio.all_tasks()
-    assert not any(t.get_name().startswith("dsm_cleaner_task") for t in all_tasks)
+    assert not any(
+        t.get_name().startswith(
+            f"exclusive_task_starter_{_TASK_NAME_PERIODICALY_CLEAN_DSM}"
+        )
+        for t in all_tasks
+    )
 
 
 async def test_dsm_cleaner_task_restarts_if_error(