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

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

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()