Merge branch 'master' into buy-credits

.env-devel

@@ -216,7 +216,7 @@ DIRECTOR_PORT=8080
 
 DIRECTOR_V2_HOST=director-v2
 DIRECTOR_V2_PORT=8000
-DIRECTOR_V2_DYNAMIC_SCHEDULER_IGNORE_SERVICES_SHUTDOWN_WHEN_CREDITS_LIMIT_REACHED=1
+DIRECTOR_V2_DYNAMIC_SCHEDULER_CLOSE_SERVICES_VIA_FRONTEND_WHEN_CREDITS_LIMIT_REACHED=1
 DIRECTOR_V2_SERVICES_CUSTOM_CONSTRAINTS=[]
 DIRECTOR_V2_GENERIC_RESOURCE_PLACEMENT_CONSTRAINTS_SUBSTITUTIONS='{}'
 DIRECTOR_V2_NODE_PORTS_STORAGE_AUTH=null

packages/models-library/src/models_library/api_schemas_directorv2/notifications.py

@@ -0,0 +1,8 @@
+from models_library.projects_nodes_io import NodeID
+from models_library.wallets import WalletID
+from pydantic import BaseModel
+
+
+class ServiceNoMoreCredits(BaseModel):
+    node_id: NodeID
+    wallet_id: WalletID

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

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

services/api-server/src/simcore_service_api_server/api/errors/httpx_client_error.py

@@ -6,7 +6,8 @@
 import logging
 from typing import Any
 
-from fastapi import HTTPException, Request, status
+from fastapi import Request, status
+from fastapi.responses import JSONResponse
 from httpx import HTTPError, TimeoutException
 
 _logger = logging.getLogger(__file__)
@@ -30,6 +31,6 @@ async def handle_httpx_client_exceptions(_: Request, exc: HTTPError):
 
     if status_code >= status.HTTP_500_INTERNAL_SERVER_ERROR:
         _logger.exception("%s. host=%s. %s", detail, exc.request.url.host, f"{exc}")
-    raise HTTPException(
-        status_code=status_code, detail=detail, headers=headers
-    ) from exc
+    return JSONResponse(
+        status_code=status_code, content={"detail": detail}, headers=headers
+    )

services/director-v2/requirements/_base.in

@@ -29,6 +29,7 @@ networkx
 ordered-set
 orjson
 pydantic[dotenv]
+python-socketio
 redis
 rich
 tenacity

services/director-v2/requirements/_base.txt

@@ -96,6 +96,8 @@ attrs==23.1.0
     #   aiohttp
     #   jsonschema
     #   referencing
+bidict==0.23.1
+    # via python-socketio
 blosc==1.11.1
     # via -r requirements/../../../services/dask-sidecar/requirements/_dask-distributed.txt
 certifi==2023.7.22
@@ -197,6 +199,7 @@ h11==0.14.0
     # via
     #   httpcore
     #   uvicorn
+    #   wsproto
 httpcore==0.17.3
     # via
     #   dnspython
@@ -421,8 +424,12 @@ python-dotenv==1.0.0
     # via
     #   pydantic
     #   uvicorn
+python-engineio==4.9.0
+    # via python-socketio
 python-multipart==0.0.6
     # via fastapi
+python-socketio==5.11.1
+    # via -r requirements/_base.in
 pyyaml==6.0.1
     # via
     #   -c requirements/../../../packages/dask-task-models-library/requirements/../../../packages/models-library/requirements/../../../requirements/constraints.txt
@@ -501,6 +508,8 @@ rpds-py==0.9.2
     # via
     #   jsonschema
     #   referencing
+simple-websocket==1.0.0
+    # via python-engineio
 six==1.16.0
     # via python-dateutil
 sniffio==1.3.0
@@ -668,6 +677,8 @@ watchfiles==0.19.0
     # via uvicorn
 websockets==11.0.3
     # via uvicorn
+wsproto==1.2.0
+    # via simple-websocket
 yarl==1.9.2
     # via
     #   -r requirements/../../../packages/postgres-database/requirements/_base.in

services/director-v2/src/simcore_service_director_v2/core/application.py

@@ -27,9 +27,11 @@
     director_v0,
     dynamic_services,
     dynamic_sidecar,
+    notifier,
     osparc_variables_substitutions,
     rabbitmq,
     resource_usage_tracker_client,
+    socketio,
     storage,
 )
 from .errors import (
@@ -165,6 +167,8 @@ def init_app(settings: AppSettings | None = None) -> FastAPI:
     if dynamic_scheduler_enabled:
         dynamic_sidecar.setup(app)
         api_keys_manager.setup(app)
+        socketio.setup(app)
+        notifier.setup(app)
 
     if (
         settings.DIRECTOR_V2_COMPUTATIONAL_BACKEND.COMPUTATIONAL_BACKEND_DASK_CLIENT_ENABLED
@@ -174,8 +178,7 @@ def init_app(settings: AppSettings | None = None) -> FastAPI:
     if computational_backend_enabled:
         comp_scheduler.setup(app)
 
-    if settings.DIRECTOR_V2_RESOURCE_USAGE_TRACKER:
-        resource_usage_tracker_client.setup(app)
+    resource_usage_tracker_client.setup(app)
 
     if settings.DIRECTOR_V2_PROMETHEUS_INSTRUMENTATION_ENABLED:
         setup_prometheus_instrumentation(app)

services/director-v2/src/simcore_service_director_v2/core/dynamic_services_settings/scheduler.py

@@ -1,7 +1,7 @@
 from typing import Final
 
 from models_library.projects_networks import DockerNetworkName
-from pydantic import Field, NonNegativeInt, PositiveFloat, PositiveInt
+from pydantic import Field, NonNegativeInt, PositiveFloat
 from settings_library.base import BaseCustomSettings
 
 _MINUTE: Final[NonNegativeInt] = 60
@@ -59,6 +59,14 @@ class DynamicServicesSchedulerSettings(BaseCustomSettings):
         description="Prometheus will scrape service placed on these networks",
     )
 
+    DIRECTOR_V2_DYNAMIC_SCHEDULER_CLOSE_SERVICES_VIA_FRONTEND_WHEN_CREDITS_LIMIT_REACHED: bool = Field(
+        default=True,
+        description=(
+            "when the message indicating there are no more credits left in a wallet "
+            "the director-v2 will shutdown the services via the help of the frontend"
+        ),
+    )
+
     #
     # TIMEOUTS AND RETRY dark worlds
     #
@@ -139,19 +147,3 @@ class DynamicServicesSchedulerSettings(BaseCustomSettings):
             "allow for some time to pass before declaring it failed."
         ),
     )
-
-    #
-    # DEVELOPMENT ONLY config
-    #
-
-    DIRECTOR_V2_DYNAMIC_SCHEDULER_IGNORE_SERVICES_SHUTDOWN_WHEN_CREDITS_LIMIT_REACHED: bool = Field(
-        default=True,
-        description=(
-            "when the message indicating there are no more credits left in a wallet "
-            "the director-v2 will shutdown services if True"
-        ),
-    )
-
-    DYNAMIC_SIDECAR_DOCKER_NODE_CONCURRENT_RESOURCE_SLOTS: PositiveInt = Field(
-        2, description="Amount of slots per resource on a node"
-    )

services/director-v2/src/simcore_service_director_v2/modules/notifier.py

@@ -0,0 +1,59 @@
+import contextlib
+
+import socketio
+from fastapi import FastAPI
+from fastapi.encoders import jsonable_encoder
+from models_library.api_schemas_directorv2.notifications import ServiceNoMoreCredits
+from models_library.api_schemas_directorv2.socketio import (
+    SOCKET_IO_SERVICE_NO_MORE_CREDITS_EVENT,
+)
+from models_library.api_schemas_webserver.socketio import SocketIORoomStr
+from models_library.projects_nodes_io import NodeID
+from models_library.users import UserID
+from models_library.wallets import WalletID
+from servicelib.fastapi.app_state import SingletonInAppStateMixin
+
+
+class Notifier(SingletonInAppStateMixin):
+    app_state_name: str = "notifier"
+
+    def __init__(self, sio_manager: socketio.AsyncAioPikaManager):
+        self._sio_manager = sio_manager
+
+    async def notify_shutdown_no_more_credits(
+        self, user_id: UserID, node_id: NodeID, wallet_id: WalletID
+    ) -> None:
+        await self._sio_manager.emit(
+            SOCKET_IO_SERVICE_NO_MORE_CREDITS_EVENT,
+            data=jsonable_encoder(
+                ServiceNoMoreCredits(node_id=node_id, wallet_id=wallet_id)
+            ),
+            room=SocketIORoomStr.from_user_id(user_id),
+        )
+
+
+async def publish_shutdown_no_more_credits(
+    app: FastAPI, *, user_id: UserID, node_id: NodeID, wallet_id: WalletID
+) -> None:
+    notifier: Notifier = Notifier.get_from_app_state(app)
+    await notifier.notify_shutdown_no_more_credits(
+        user_id=user_id, node_id=node_id, wallet_id=wallet_id
+    )
+
+
+def setup(app: FastAPI):
+    async def _on_startup() -> None:
+        assert app.state.external_socketio  # nosec
+
+        notifier = Notifier(
+            sio_manager=app.state.external_socketio,
+        )
+        notifier.set_to_app_state(app)
+        assert Notifier.get_from_app_state(app) == notifier  # nosec
+
+    async def _on_shutdown() -> None:
+        with contextlib.suppress(AttributeError):
+            Notifier.pop_from_app_state(app)
+
+    app.add_event_handler("startup", _on_startup)
+    app.add_event_handler("shutdown", _on_shutdown)

services/director-v2/src/simcore_service_director_v2/modules/rabbitmq.py

@@ -16,6 +16,7 @@
 
 from ..core.errors import ConfigurationError
 from ..core.settings import AppSettings
+from .notifier import publish_shutdown_no_more_credits
 
 _logger = logging.getLogger(__name__)
 
@@ -27,9 +28,20 @@ async def handler_out_of_credits(app: FastAPI, data: bytes) -> bool:
     settings: AppSettings = app.state.settings
 
     if (
-        settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER.DIRECTOR_V2_DYNAMIC_SCHEDULER_IGNORE_SERVICES_SHUTDOWN_WHEN_CREDITS_LIMIT_REACHED
+        settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER.DIRECTOR_V2_DYNAMIC_SCHEDULER_CLOSE_SERVICES_VIA_FRONTEND_WHEN_CREDITS_LIMIT_REACHED
     ):
-        _logger.debug("Skipped shutting down services for wallet %s", message.wallet_id)
+        _logger.warning(
+            "Notifying frontend to shutdown service: '%s' for user '%s' because wallet '%s' is out of credits.",
+            message.node_id,
+            message.user_id,
+            message.wallet_id,
+        )
+        await publish_shutdown_no_more_credits(
+            app,
+            user_id=message.user_id,
+            node_id=message.node_id,
+            wallet_id=message.wallet_id,
+        )
     else:
         await scheduler.mark_all_services_in_wallet_for_removal(
             wallet_id=message.wallet_id

services/director-v2/src/simcore_service_director_v2/modules/socketio.py

@@ -0,0 +1,33 @@
+import logging
+
+import socketio
+from fastapi import FastAPI
+from servicelib.socketio_utils import cleanup_socketio_async_pubsub_manager
+
+from ..core.settings import AppSettings
+
+_logger = logging.getLogger(__name__)
+
+
+def setup(app: FastAPI):
+    settings: AppSettings = app.state.settings
+
+    async def _on_startup() -> None:
+        assert app.state.rabbitmq_client  # nosec
+
+        # Connect to the as an external process in write-only mode
+        # SEE https://python-socketio.readthedocs.io/en/stable/server.html#emitting-from-external-processes
+        app.state.external_socketio = socketio.AsyncAioPikaManager(
+            url=settings.DIRECTOR_V2_RABBITMQ.dsn,
+            logger=_logger,
+            write_only=True,
+        )
+
+    async def _on_shutdown() -> None:
+        if external_socketio := getattr(app.state, "external_socketio"):  # noqa: B009
+            await cleanup_socketio_async_pubsub_manager(
+                server_manager=external_socketio
+            )
+
+    app.add_event_handler("startup", _on_startup)
+    app.add_event_handler("shutdown", _on_shutdown)

services/director-v2/tests/conftest.py

@@ -38,6 +38,7 @@
     "pytest_simcore.postgres_service",
     "pytest_simcore.pydantic_models",
     "pytest_simcore.pytest_global_environs",
+    "pytest_simcore.pytest_socketio",
     "pytest_simcore.rabbit_service",
     "pytest_simcore.redis_service",
     "pytest_simcore.repository_paths",
@@ -237,7 +238,7 @@ def fake_workbench_as_dict(fake_workbench_file: Path) -> dict[str, Any]:
 
 @pytest.fixture
 def fake_workbench_without_outputs(
-    fake_workbench_as_dict: dict[str, Any]
+    fake_workbench_as_dict: dict[str, Any],
 ) -> dict[str, Any]:
     workbench = deepcopy(fake_workbench_as_dict)
     # remove all the outputs from the workbench