🐛Garbage collector: possible fix to "spurious" shutdowns (#5768)

Author: sanderegg

packages/postgres-database/requirements/_base.in

@@ -5,5 +5,6 @@
 --constraint ./constraints.txt
 
 alembic
+pydantic
 sqlalchemy[postgresql_psycopg2binary,postgresql_asyncpg] # SEE extras in https://github.com/sqlalchemy/sqlalchemy/blob/main/setup.cfg#L43
 yarl

packages/postgres-database/requirements/_base.txt

@@ -13,8 +13,11 @@ multidict==6.0.4
     # via yarl
 psycopg2-binary==2.9.9
     # via sqlalchemy
+pydantic==1.10.15
 sqlalchemy==1.4.50
     # via alembic
 typing-extensions==4.8.0
-    # via alembic
+    # via
+    #   alembic
+    #   pydantic
 yarl==1.9.2

packages/postgres-database/src/simcore_postgres_database/utils_projects_nodes.py

@@ -1,18 +1,16 @@
 import datetime
 import uuid
-from dataclasses import asdict, dataclass, field, fields
+from dataclasses import dataclass
 from typing import Any
 
 import sqlalchemy
 from aiopg.sa.connection import SAConnection
-from simcore_postgres_database.models.projects_node_to_pricing_unit import (
-    projects_node_to_pricing_unit,
-)
+from pydantic import BaseModel, Field
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from .errors import ForeignKeyViolation, UniqueViolation
+from .models.projects_node_to_pricing_unit import projects_node_to_pricing_unit
 from .models.projects_nodes import projects_nodes
-from .utils_models import FromRowMixin
 
 
 #
@@ -38,23 +36,27 @@ class ProjectNodesDuplicateNode(BaseProjectNodesError):
     ...
 
 
-@dataclass(frozen=True, slots=True, kw_only=True)
-class ProjectNodeCreate:
+class ProjectNodeCreate(BaseModel):
     node_id: uuid.UUID
-    required_resources: dict[str, Any] = field(default_factory=dict)
+    required_resources: dict[str, Any] = Field(default_factory=dict)
 
-    @staticmethod
-    def get_field_names(*, exclude: set[str]) -> set[str]:
-        return {f.name for f in fields(ProjectNodeCreate) if f.name not in exclude}
+    @classmethod
+    def get_field_names(cls, *, exclude: set[str]) -> set[str]:
+        return {name for name in cls.__fields__ if name not in exclude}
+
+    class Config:
+        frozen = True
 
 
-@dataclass(frozen=True, slots=True, kw_only=True)
-class ProjectNode(ProjectNodeCreate, FromRowMixin):
+class ProjectNode(ProjectNodeCreate):
     created: datetime.datetime
     modified: datetime.datetime
 
+    class Config(ProjectNodeCreate.Config):
+        orm_mode = True
+
 
-@dataclass(frozen=True, slots=True, kw_only=True)
+@dataclass(frozen=True, kw_only=True)
 class ProjectNodesRepo:
     project_uuid: uuid.UUID
 
@@ -82,7 +84,7 @@ async def add(
                 [
                     {
                         "project_uuid": f"{self.project_uuid}",
-                        **asdict(node),
+                        **node.dict(),
                     }
                     for node in nodes
                 ]
@@ -101,7 +103,7 @@ async def add(
             assert result  # nosec
             rows = await result.fetchall()
             assert rows is not None  # nosec
-            return [ProjectNode.from_row(r) for r in rows]
+            return [ProjectNode.from_orm(r) for r in rows]
         except ForeignKeyViolation as exc:
             # this happens when the project does not exist, as we first check the node exists
             msg = f"Project {self.project_uuid} not found"
@@ -127,7 +129,7 @@ async def list(self, connection: SAConnection) -> list[ProjectNode]:  # noqa: A0
         assert result  # nosec
         rows = await result.fetchall()
         assert rows is not None  # nosec
-        return [ProjectNode.from_row(row) for row in rows]
+        return [ProjectNode.from_orm(row) for row in rows]
 
     async def get(self, connection: SAConnection, *, node_id: uuid.UUID) -> ProjectNode:
         """get a node in the current project
@@ -152,7 +154,7 @@ async def get(self, connection: SAConnection, *, node_id: uuid.UUID) -> ProjectN
             msg = f"Node with {node_id} not found"
             raise ProjectNodesNodeNotFound(msg)
         assert row  # nosec
-        return ProjectNode.from_row(row)
+        return ProjectNode.from_orm(row)
 
     async def update(
         self, connection: SAConnection, *, node_id: uuid.UUID, **values
@@ -181,7 +183,7 @@ async def update(
             msg = f"Node with {node_id} not found"
             raise ProjectNodesNodeNotFound(msg)
         assert row  # nosec
-        return ProjectNode.from_row(row)
+        return ProjectNode.from_orm(row)
 
     async def delete(self, connection: SAConnection, *, node_id: uuid.UUID) -> None:
         """delete a node in the current project

services/web/server/src/simcore_service_webserver/director_v2/_core_dynamic_services.py

@@ -7,9 +7,10 @@
 import logging
 
 from aiohttp import web
+from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
 from models_library.projects import ProjectID
 from models_library.services import ServicePortKey
-from pydantic import BaseModel
+from pydantic import BaseModel, parse_obj_as
 from pydantic.types import NonNegativeFloat, PositiveInt
 from servicelib.logging_utils import log_decorator
 from yarl import URL
@@ -30,7 +31,7 @@ async def list_dynamic_services(
     app: web.Application,
     user_id: PositiveInt | None = None,
     project_id: str | None = None,
-) -> list[DataType]:
+) -> list[DynamicServiceGet]:
     params = _Params(user_id=user_id, project_id=project_id)
     params_dict = params.dict(exclude_none=True)
     settings: DirectorV2Settings = get_plugin_settings(app)
@@ -48,7 +49,7 @@ async def list_dynamic_services(
     if services is None:
         services = []
     assert isinstance(services, list)  # nosec
-    return services
+    return parse_obj_as(list[DynamicServiceGet], services)
 
 
 # NOTE: ANE https://github.com/ITISFoundation/osparc-simcore/issues/3191

services/web/server/src/simcore_service_webserver/dynamic_scheduler/api.py

@@ -118,11 +118,11 @@ async def stop_dynamic_services_in_project(
         services_to_stop = [
             stop_dynamic_service(
                 app=app,
-                node_id=service["service_uuid"],
+                node_id=service.node_uuid,
                 simcore_user_agent=simcore_user_agent,
                 save_state=save_state,
                 progress=progress_bar.sub_progress(
-                    1, description=service["service_uuid"]
+                    1, description=f"{service.node_uuid}"
                 ),
             )
             for service in running_dynamic_services

services/web/server/src/simcore_service_webserver/garbage_collector/_core.py

@@ -58,8 +58,4 @@ async def collect_garbage(app: web.Application):
         # the projects are closed or the user was disconencted.
         # This will close and remove all these services from
         # the cluster, thus freeing important resources.
-
-        # Temporary disabling GC to until the dynamic service
-        # safe function is invoked by the GC. This will avoid
-        # data loss for current users.
         await remove_orphaned_services(registry, app)