✨ dynamic-scheduler pushes service state to the frontend (⚠️ devops)

packages/models-library/src/models_library/api_schemas_dynamic_scheduler/socketio.py

@@ -0,0 +1,3 @@
+from typing import Final
+
+SOCKET_IO_SERVICE_STATUS_EVENT: Final[str] = "serviceStatus"

packages/models-library/src/models_library/api_schemas_webserver/projects_nodes.py

@@ -93,19 +93,36 @@ class NodeGet(OutputSchema):
 
     class Config:
         schema_extra: ClassVar[dict[str, Any]] = {
-            "example": {
-                "published_port": 30000,
-                "entrypoint": "/the/entry/point/is/here",
-                "service_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
-                "service_key": "simcore/services/comp/itis/sleeper",
-                "service_version": "1.2.3",
-                "service_host": "jupyter_E1O2E-LAH",
-                "service_port": 8081,
-                "service_basepath": "/x/E1O2E-LAH",
-                "service_state": "pending",
-                "service_message": "no suitable node (insufficient resources on 1 node)",
-                "user_id": 123,
-            }
+            "examples": [
+                # computational
+                {
+                    "published_port": 30000,
+                    "entrypoint": "/the/entry/point/is/here",
+                    "service_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+                    "service_key": "simcore/services/comp/itis/sleeper",
+                    "service_version": "1.2.3",
+                    "service_host": "jupyter_E1O2E-LAH",
+                    "service_port": 8081,
+                    "service_basepath": "/x/E1O2E-LAH",
+                    "service_state": "pending",
+                    "service_message": "no suitable node (insufficient resources on 1 node)",
+                    "user_id": 123,
+                },
+                # dynamic
+                {
+                    "published_port": 30000,
+                    "entrypoint": "/the/entry/point/is/here",
+                    "service_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+                    "service_key": "simcore/services/dynamic/some-dynamic-service",
+                    "service_version": "1.2.3",
+                    "service_host": "jupyter_E1O2E-LAH",
+                    "service_port": 8081,
+                    "service_basepath": "/x/E1O2E-LAH",
+                    "service_state": "pending",
+                    "service_message": "no suitable node (insufficient resources on 1 node)",
+                    "user_id": 123,
+                },
+            ]
         }
 
 

packages/models-library/src/models_library/services_enums.py

@@ -11,14 +11,18 @@ class ServiceBootType(str, Enum):
 @functools.total_ordering
 @unique
 class ServiceState(Enum):
+    FAILED = "failed"
+
     PENDING = "pending"
     PULLING = "pulling"
     STARTING = "starting"
     RUNNING = "running"
-    COMPLETE = "complete"
-    FAILED = "failed"
+
     STOPPING = "stopping"
 
+    COMPLETE = "complete"
+    IDLE = "idle"
+
     def __lt__(self, other):
         if self.__class__ is other.__class__:
             comparison_order = ServiceState.comparison_order()
@@ -39,6 +43,7 @@ def comparison_order() -> dict["ServiceState", int]:
             ServiceState.RUNNING: 4,
             ServiceState.STOPPING: 5,
             ServiceState.COMPLETE: 6,
+            ServiceState.IDLE: 7,
         }
 
 

packages/models-library/tests/conftest.py

@@ -9,6 +9,7 @@
 import pytest
 
 pytest_plugins = [
+    "pytest_simcore.faker_projects_data",
     "pytest_simcore.pydantic_models",
     "pytest_simcore.pytest_global_environs",
     "pytest_simcore.repository_paths",

packages/models-library/tests/test_utils_nodes.py

@@ -16,12 +16,6 @@
 from models_library.utils.nodes import compute_node_hash
 from pydantic import AnyUrl, parse_obj_as
 
-
-@pytest.fixture()
-def node_id() -> NodeID:
-    return uuid4()
-
-
 ANOTHER_NODE_ID = uuid4()
 ANOTHER_NODE_OUTPUT_KEY = "the_output_link"
 ANOTHER_NODE_PAYLOAD = {"outputs": {ANOTHER_NODE_OUTPUT_KEY: 36}}

packages/service-library/src/servicelib/deferred_tasks/_redis_task_tracker.py

@@ -1,3 +1,4 @@
+import pickle
 from typing import Final
 from uuid import uuid4
 
@@ -33,13 +34,15 @@ async def get_new_unique_identifier(self) -> TaskUID:
 
     async def _get_raw(self, redis_key: str) -> TaskScheduleModel | None:
         found_data = await self.redis_client_sdk.redis.get(redis_key)
-        return None if found_data is None else TaskScheduleModel.parse_raw(found_data)
+        return None if found_data is None else pickle.loads(found_data)  # noqa: S301
 
     async def get(self, task_uid: TaskUID) -> TaskScheduleModel | None:
         return await self._get_raw(_get_key(task_uid))
 
     async def save(self, task_uid: TaskUID, task_schedule: TaskScheduleModel) -> None:
-        await self.redis_client_sdk.redis.set(_get_key(task_uid), task_schedule.json())
+        await self.redis_client_sdk.redis.set(
+            _get_key(task_uid), pickle.dumps(task_schedule)
+        )
 
     async def remove(self, task_uid: TaskUID) -> None:
         await self.redis_client_sdk.redis.delete(_get_key(task_uid))

packages/service-library/src/servicelib/services_utils.py

@@ -1,5 +1,11 @@
 import urllib.parse
 
+from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_webserver.projects_nodes import (
+    NodeGet,
+    NodeGetIdle,
+    NodeGetUnknown,
+)
 from models_library.services import ServiceType
 
 
@@ -9,3 +15,14 @@ def get_service_from_key(service_key: str) -> ServiceType:
     if encoded_service_type == "comp":
         encoded_service_type = "computational"
     return ServiceType(encoded_service_type)
+
+
+def get_status_as_dict(
+    status: NodeGetIdle | NodeGetUnknown | DynamicServiceGet | NodeGet,
+) -> dict:
+    """shared between different backend services to guarantee same result to frontend"""
+    return (
+        status.dict(by_alias=True)
+        if isinstance(status, DynamicServiceGet)
+        else status.dict()
+    )

packages/service-library/tests/conftest.py

@@ -76,9 +76,11 @@ async def get_redis_client_sdk(
     Callable[[RedisDatabase], AbstractAsyncContextManager[RedisClientSDK]]
 ]:
     @asynccontextmanager
-    async def _(database: RedisDatabase) -> AsyncIterator[RedisClientSDK]:
+    async def _(
+        database: RedisDatabase, decode_response: bool = True  # noqa: FBT002
+    ) -> AsyncIterator[RedisClientSDK]:
         redis_resources_dns = redis_service.build_redis_dsn(database)
-        client = RedisClientSDK(redis_resources_dns)
+        client = RedisClientSDK(redis_resources_dns, decode_responses=decode_response)
         assert client
         assert client.redis_dsn == redis_resources_dns
         await client.setup()

packages/service-library/tests/deferred_tasks/conftest.py

@@ -9,8 +9,10 @@
 @pytest.fixture
 async def redis_client_sdk_deferred_tasks(
     get_redis_client_sdk: Callable[
-        [RedisDatabase], AbstractAsyncContextManager[RedisClientSDK]
+        [RedisDatabase, bool], AbstractAsyncContextManager[RedisClientSDK]
     ]
 ) -> AsyncIterator[RedisClientSDK]:
-    async with get_redis_client_sdk(RedisDatabase.DEFERRED_TASKS) as client:
+    async with get_redis_client_sdk(
+        RedisDatabase.DEFERRED_TASKS, decode_response=False
+    ) as client:
         yield client

packages/service-library/tests/deferred_tasks/example_app.py

@@ -8,6 +8,7 @@
 from uuid import uuid4
 
 from pydantic import NonNegativeInt
+from redis.asyncio import Redis
 from servicelib.deferred_tasks import (
     BaseDeferredHandler,
     DeferredContext,
@@ -54,21 +55,22 @@ async def on_result(cls, result: str, context: DeferredContext) -> None:
 
 class InMemoryLists:
     def __init__(self, redis_settings: RedisSettings, port: int) -> None:
-        self.redis_client_sdk = RedisClientSDK(
-            redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS)
-        )
+        # NOTE: RedisClientSDK is not required here but it's used to easily construct
+        # a redis connection
+        self.redis: Redis = RedisClientSDK(
+            redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
+            decode_responses=True,
+        ).redis
         self.port = port
 
     def _get_queue_name(self, queue_name: str) -> str:
         return f"in_memory_lists::{queue_name}.{self.port}"
 
     async def append_to(self, queue_name: str, value: Any) -> None:
-        await self.redis_client_sdk.redis.rpush(self._get_queue_name(queue_name), value)  # type: ignore
+        await self.redis.rpush(self._get_queue_name(queue_name), value)  # type: ignore
 
     async def get_all_from(self, queue_name: str) -> list:
-        return await self.redis_client_sdk.redis.lrange(
-            self._get_queue_name(queue_name), 0, -1
-        )  # type: ignore
+        return await self.redis.lrange(self._get_queue_name(queue_name), 0, -1)  # type: ignore
 
 
 class ExampleApp:
@@ -79,18 +81,19 @@ def __init__(
         in_memory_lists: InMemoryLists,
         max_workers: NonNegativeInt,
     ) -> None:
-        self._redis_client_sdk = RedisClientSDK(
-            redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS)
+        self._redis_client = RedisClientSDK(
+            redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
+            decode_responses=False,
         )
         self._manager = DeferredManager(
             rabbit_settings,
-            self._redis_client_sdk,
+            self._redis_client,
             globals_context={"in_memory_lists": in_memory_lists},
             max_workers=max_workers,
         )
 
     async def setup(self) -> None:
-        await self._redis_client_sdk.setup()
+        await self._redis_client.setup()
         await self._manager.setup()
 
 

packages/service-library/tests/deferred_tasks/test__base_deferred_handler.py

@@ -52,7 +52,10 @@ class MockKeys(StrAutoEnum):
 async def redis_client_sdk(
     redis_service: RedisSettings,
 ) -> AsyncIterable[RedisClientSDK]:
-    sdk = RedisClientSDK(redis_service.build_redis_dsn(RedisDatabase.DEFERRED_TASKS))
+    sdk = RedisClientSDK(
+        redis_service.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
+        decode_responses=False,
+    )
     await sdk.setup()
     yield sdk
     await sdk.shutdown()

packages/settings-library/src/settings_library/redis.py

@@ -17,6 +17,7 @@ class RedisDatabase(IntEnum):
     ANNOUNCEMENTS = 5
     DISTRIBUTED_IDENTIFIERS = 6
     DEFERRED_TASKS = 7
+    DYNAMIC_SERVICES = 8
 
 
 class RedisSettings(BaseCustomSettings):

services/director-v2/tests/conftest.py

@@ -42,19 +42,20 @@
     "pytest_simcore.docker_registry",
     "pytest_simcore.docker_swarm",
     "pytest_simcore.environment_configs",
+    "pytest_simcore.faker_projects_data",
     "pytest_simcore.faker_users_data",
     "pytest_simcore.minio_service",
     "pytest_simcore.postgres_service",
     "pytest_simcore.pydantic_models",
     "pytest_simcore.pytest_global_environs",
-    "pytest_simcore.socketio",
     "pytest_simcore.rabbit_service",
     "pytest_simcore.redis_service",
     "pytest_simcore.repository_paths",
     "pytest_simcore.schemas",
     "pytest_simcore.simcore_dask_service",
     "pytest_simcore.simcore_services",
     "pytest_simcore.simcore_storage_service",
+    "pytest_simcore.socketio",
 ]
 
 logger = logging.getLogger(__name__)

services/director-v2/tests/unit/test_modules_dask_client.py

@@ -284,11 +284,6 @@ def project_id() -> ProjectID:
     return uuid4()
 
 
-@pytest.fixture
-def node_id() -> NodeID:
-    return uuid4()
-
-
 @dataclass
 class ImageParams:
     image: Image

services/docker-compose-ops.yml

@@ -93,7 +93,8 @@ services:
         user_notifications:${REDIS_HOST}:${REDIS_PORT}:4:${REDIS_PASSWORD},
         announcements:${REDIS_HOST}:${REDIS_PORT}:5:${REDIS_PASSWORD},
         distributed_identifiers:${REDIS_HOST}:${REDIS_PORT}:6:${REDIS_PASSWORD},
-        deferred_tasks:${REDIS_HOST}:${REDIS_PORT}:7:${REDIS_PASSWORD}
+        deferred_tasks:${REDIS_HOST}:${REDIS_PORT}:7:${REDIS_PASSWORD},
+        dynamic_services:${REDIS_HOST}:${REDIS_PORT}:8:${REDIS_PASSWORD}
       # If you add/remove a db, do not forget to update the --databases entry in the docker-compose.yml
     ports:
       - "18081:8081"

services/docker-compose.yml

@@ -1168,7 +1168,19 @@ services:
       # also aof (append only) is also enabled such that we get full durability at the expense
       # of backup size. The backup is written into /data.
       # https://redis.io/topics/persistence
-      [ "redis-server", "--save", "60 1", "--loglevel", "verbose", "--databases", "8", "--appendonly", "yes", "--requirepass", "${REDIS_PASSWORD}" ]
+      [
+        "redis-server",
+        "--save",
+        "60 1",
+        "--loglevel",
+        "verbose",
+        "--databases",
+        "9",
+        "--appendonly",
+        "yes",
+        "--requirepass",
+        "${REDIS_PASSWORD}"
+      ]
     networks:
       - default
       - autoscaling_subnet

services/dynamic-scheduler/requirements/_base.in

@@ -14,9 +14,10 @@
 --requirement ../../../packages/service-library/requirements/_fastapi.in
 
 
-
+arrow
 fastapi
 httpx
 packaging
+python-socketio
 typer[all]
 uvicorn[standard]

services/dynamic-scheduler/requirements/_base.txt

@@ -47,6 +47,8 @@ attrs==23.2.0
     #   aiohttp
     #   jsonschema
     #   referencing
+bidict==0.23.1
+    # via python-socketio
 certifi==2024.2.2
     # via
     #   -c requirements/../../../packages/models-library/requirements/../../../requirements/constraints.txt
@@ -107,6 +109,7 @@ h11==0.14.0
     # via
     #   httpcore
     #   uvicorn
+    #   wsproto
 httpcore==1.0.5
     # via httpx
 httptools==0.6.1
@@ -265,6 +268,10 @@ python-dateutil==2.9.0.post0
     # via arrow
 python-dotenv==1.0.1
     # via uvicorn
+python-engineio==4.9.1
+    # via python-socketio
+python-socketio==5.11.2
+    # via -r requirements/_base.in
 pyyaml==6.0.1
     # via
     #   -c requirements/../../../packages/models-library/requirements/../../../requirements/constraints.txt
@@ -306,6 +313,8 @@ setuptools==74.0.0
     # via opentelemetry-instrumentation
 shellingham==1.5.4
     # via typer
+simple-websocket==1.0.0
+    # via python-engineio
 six==1.16.0
     # via python-dateutil
 sniffio==1.3.1

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/rest/_dependencies.py

@@ -3,7 +3,8 @@
 from servicelib.fastapi.dependencies import get_app, get_reverse_url_mapper
 from servicelib.rabbitmq import RabbitMQClient, RabbitMQRPCClient
 from servicelib.redis import RedisClientSDK
-from simcore_service_dynamic_scheduler.services.redis import get_redis_client
+from settings_library.redis import RedisDatabase
+from simcore_service_dynamic_scheduler.services.redis import get_all_redis_clients
 
 from ...services.rabbitmq import get_rabbitmq_client, get_rabbitmq_rpc_server
 
@@ -19,8 +20,10 @@ def get_rabbitmq_rpc_server_from_request(request: Request) -> RabbitMQRPCClient:
     return get_rabbitmq_rpc_server(request.app)
 
 
-def get_redis_client_from_request(request: Request) -> RedisClientSDK:
-    return get_redis_client(request.app)
+def get_redis_clients_from_request(
+    request: Request,
+) -> dict[RedisDatabase, RedisClientSDK]:
+    return get_all_redis_clients(request.app)
 
 
 __all__: tuple[str, ...] = (