✨ dynamic-sidecar actively monitors disk usage ⚠️ (#5248)

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

README.md

@@ -1,7 +1,9 @@
 # osparc-simcore platform
 
 <p align="center">
+<a href="https://osparc.io" target="_blank">
 <img src="https://user-images.githubusercontent.com/32800795/61083844-ff48fb00-a42c-11e9-8e63-fa2d709c8baf.png" width="500">
+</a>
 </p>
 
 <!-- BADGES: LINKS ON CLICK --------------------------------------------------------------->
@@ -184,7 +186,9 @@ This project is licensed under the terms of the [MIT license](LICENSE).
 ---
 
 <p align="center">
+<a href="https://www.z43.swiss" target="_blank">
 <image src="https://raw.githubusercontent.com/ITISFoundation/osparc-simcore-clients/4e8b18494f3191d55f6692a6a605818aeeb83f95/docs/_media/mwl.png" alt="Made with love (and lots of hard work) at www.z43.swiss" width="20%" />
+</a>
 </p>
 
 <!-- ADD REFERENCES BELOW AND KEEP THEM IN ALPHABETICAL ORDER -->

packages/models-library/requirements/_test.in

@@ -11,6 +11,7 @@
 coverage
 faker
 pint
+psutil
 pytest
 pytest-aiohttp  # incompatible with pytest-asyncio. See https://github.com/pytest-dev/pytest-asyncio/issues/76
 pytest-cov

packages/models-library/requirements/_test.txt

@@ -52,6 +52,8 @@ pluggy==1.3.0
     # via pytest
 pprintpp==0.4.0
     # via pytest-icdiff
+psutil==5.9.8
+    # via -r requirements/_test.in
 pytest==7.4.3
     # via
     #   -r requirements/_test.in

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

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

packages/models-library/src/models_library/api_schemas_dynamic_sidecar/telemetry.py

@@ -0,0 +1,58 @@
+from abc import abstractmethod
+from pathlib import Path
+from typing import Protocol
+
+from models_library.projects_nodes_io import NodeID
+from pydantic import BaseModel, ByteSize, Field
+
+
+class SDiskUsageProtocol(Protocol):
+    @property
+    @abstractmethod
+    def total(self) -> int:
+        ...
+
+    @property
+    @abstractmethod
+    def used(self) -> int:
+        ...
+
+    @property
+    @abstractmethod
+    def free(self) -> int:
+        ...
+
+    @property
+    @abstractmethod
+    def percent(self) -> float:
+        ...
+
+
+class DiskUsage(BaseModel):
+    used: ByteSize = Field(description="used space")
+    free: ByteSize = Field(description="remaining space")
+
+    total: ByteSize = Field(description="total space = free + used")
+    used_percent: float = Field(
+        gte=0.00,
+        lte=100.00,
+        description="Percent of used space relative to the total space",
+    )
+
+    @classmethod
+    def from_ps_util_disk_usage(
+        cls, ps_util_disk_usage: SDiskUsageProtocol
+    ) -> "DiskUsage":
+        total = ps_util_disk_usage.free + ps_util_disk_usage.used
+        used_percent = round(ps_util_disk_usage.used * 100 / total, 2)
+        return cls(
+            used=ByteSize(ps_util_disk_usage.used),
+            free=ByteSize(ps_util_disk_usage.free),
+            total=ByteSize(total),
+            used_percent=used_percent,
+        )
+
+
+class ServiceDiskUsage(BaseModel):
+    node_id: NodeID
+    usage: dict[Path, DiskUsage]

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

@@ -1,17 +1,16 @@
+from ..basic_types import IDStr
 from ..users import GroupID, UserID
 
 
-class SocketIORoom(str):
-    __slots__ = ()
-
+class SocketIORoomStr(IDStr):
     @classmethod
-    def from_socket_id(cls, socket_id: str) -> "SocketIORoom":
+    def from_socket_id(cls, socket_id: str) -> "SocketIORoomStr":
         return cls(socket_id)
 
     @classmethod
-    def from_group_id(cls, group_id: GroupID) -> "SocketIORoom":
+    def from_group_id(cls, group_id: GroupID) -> "SocketIORoomStr":
         return cls(f"group:{group_id}")
 
     @classmethod
-    def from_user_id(cls, user_id: UserID) -> "SocketIORoom":
+    def from_user_id(cls, user_id: UserID) -> "SocketIORoomStr":
         return cls(f"user:{user_id}")

packages/models-library/tests/test_api_schemas_dynamic_sidecar_telemetry.py

@@ -0,0 +1,10 @@
+import psutil
+from models_library.api_schemas_dynamic_sidecar.telemetry import DiskUsage
+
+
+def test_disk_usage():
+    ps_util_disk_usage = psutil.disk_usage("/")
+    disk_usage = DiskUsage.from_ps_util_disk_usage(ps_util_disk_usage)
+    assert disk_usage.used == ps_util_disk_usage.used
+    assert disk_usage.free == ps_util_disk_usage.free
+    assert round(disk_usage.used_percent, 1) == round(ps_util_disk_usage.percent, 1)

packages/models-library/tests/test_api_schemas_webserver_socketio.py

@@ -2,7 +2,7 @@
 
 import pytest
 from faker import Faker
-from models_library.api_schemas_webserver.socketio import SocketIORoom
+from models_library.api_schemas_webserver.socketio import SocketIORoomStr
 from models_library.users import GroupID, UserID
 
 
@@ -22,6 +22,6 @@ def socket_id(faker: Faker) -> str:
 
 
 def test_socketio_room(user_id: UserID, group_id: GroupID, socket_id: str):
-    assert SocketIORoom.from_user_id(user_id) == f"user:{user_id}"
-    assert SocketIORoom.from_group_id(group_id) == f"group:{group_id}"
-    assert SocketIORoom.from_socket_id(socket_id) == socket_id
+    assert SocketIORoomStr.from_user_id(user_id) == f"user:{user_id}"
+    assert SocketIORoomStr.from_group_id(group_id) == f"group:{group_id}"
+    assert SocketIORoomStr.from_socket_id(socket_id) == socket_id

packages/postgres-database/src/simcore_postgres_database/migration/versions/f20f4c9fca71_added_enable_telemetry_option_to_groups_.py

@@ -0,0 +1,35 @@
+"""added enable telemetry option to groups extra properties
+
+Revision ID: f20f4c9fca71
+Revises: f9f9a650bf4b
+Create Date: 2024-01-19 14:11:16.354169+00:00
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "f20f4c9fca71"
+down_revision = "f9f9a650bf4b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "groups_extra_properties",
+        sa.Column(
+            "enable_telemetry",
+            sa.Boolean(),
+            server_default=sa.text("false"),
+            nullable=False,
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("groups_extra_properties", "enable_telemetry")
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/groups_extra_properties.py

@@ -56,6 +56,13 @@
         server_default=sa.sql.expression.false(),
         doc="If true, group will use on-demand clusters",
     ),
+    sa.Column(
+        "enable_telemetry",
+        sa.Boolean(),
+        nullable=False,
+        server_default=sa.sql.expression.false(),
+        doc="If true, will send telemetry for new style dynamic services to frontend",
+    ),
     sa.UniqueConstraint(
         "group_id", "product_name", name="group_id_product_name_uniqueness"
     ),

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

@@ -29,6 +29,7 @@ class GroupExtraProperties(FromRowMixin):
     internet_access: bool
     override_services_specifications: bool
     use_on_demand_clusters: bool
+    enable_telemetry: bool
     created: datetime.datetime
     modified: datetime.datetime
 

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

@@ -105,7 +105,7 @@ async def get_active_user_email(conn: SAConnection, user_id: int) -> str:
                 (users.c.status == UserStatus.ACTIVE) & (users.c.id == user_id)
             )
         )
-        if value:
+        if value is not None:
             assert isinstance(value, str)  # nosec
             return value
 

packages/pytest-simcore/src/pytest_simcore/pytest_socketio.py

@@ -0,0 +1,143 @@
+# pylint:disable=unused-argument
+# pylint:disable=redefined-outer-name
+
+import asyncio
+from collections.abc import AsyncIterable, AsyncIterator, Callable
+from contextlib import _AsyncGeneratorContextManager, asynccontextmanager
+from unittest.mock import AsyncMock
+
+import pytest
+import socketio
+from aiohttp import web
+from aiohttp.test_utils import TestServer
+from models_library.api_schemas_webserver.socketio import SocketIORoomStr
+from models_library.users import UserID
+from pytest_mock import MockerFixture
+from servicelib.socketio_utils import cleanup_socketio_async_pubsub_manager
+from settings_library.rabbit import RabbitSettings
+from socketio import AsyncAioPikaManager, AsyncServer
+from yarl import URL
+
+
+@pytest.fixture
+async def socketio_server_factory() -> Callable[
+    [RabbitSettings], _AsyncGeneratorContextManager[AsyncServer]
+]:
+    @asynccontextmanager
+    async def _(rabbit_settings: RabbitSettings) -> AsyncIterator[AsyncServer]:
+        # Same configuration as simcore_service_webserver/socketio/server.py
+        server_manager = AsyncAioPikaManager(url=rabbit_settings.dsn)
+
+        server = AsyncServer(
+            async_mode="aiohttp", engineio_logger=True, client_manager=server_manager
+        )
+
+        yield server
+
+        await cleanup_socketio_async_pubsub_manager(server_manager)
+
+    return _
+
+
+@pytest.fixture
+async def socketio_server() -> AsyncIterable[AsyncServer]:
+    msg = "must be implemented in test"
+    raise NotImplementedError(msg)
+
+
+@pytest.fixture
+async def web_server(
+    socketio_server: AsyncServer, unused_tcp_port_factory: Callable[[], int]
+) -> AsyncIterator[URL]:
+    """
+    this emulates the webserver setup: socketio server with
+    an aiopika manager that attaches an aiohttp web app
+    """
+    aiohttp_app = web.Application()
+    socketio_server.attach(aiohttp_app)
+
+    async def _lifespan(
+        server: TestServer, started: asyncio.Event, teardown: asyncio.Event
+    ):
+        # NOTE: this is necessary to avoid blocking comms between client and this server
+        await server.start_server()
+        started.set()  # notifies started
+        await teardown.wait()  # keeps test0server until needs to close
+        await server.close()
+
+    setup = asyncio.Event()
+    teardown = asyncio.Event()
+
+    server = TestServer(aiohttp_app, port=unused_tcp_port_factory())
+    t = asyncio.create_task(_lifespan(server, setup, teardown), name="server-lifespan")
+
+    await setup.wait()
+
+    yield URL(server.make_url("/"))
+
+    assert t
+    teardown.set()
+
+
+@pytest.fixture
+async def server_url(web_server: URL) -> str:
+    return f'{web_server.with_path("/")}'
+
+
+@pytest.fixture
+def socketio_client_factory(
+    server_url: str,
+) -> Callable[[], _AsyncGeneratorContextManager[socketio.AsyncClient]]:
+    @asynccontextmanager
+    async def _() -> AsyncIterator[socketio.AsyncClient]:
+        """This emulates a socketio client in the front-end"""
+        client = socketio.AsyncClient(logger=True, engineio_logger=True)
+        await client.connect(f"{server_url}", transports=["websocket"])
+
+        yield client
+
+        await client.disconnect()
+
+    return _
+
+
+@pytest.fixture
+def room_name() -> SocketIORoomStr:
+    msg = "must be implemented in test"
+    raise NotImplementedError(msg)
+
+
+@pytest.fixture
+def socketio_server_events(
+    socketio_server: AsyncServer,
+    mocker: MockerFixture,
+    user_id: UserID,
+    room_name: SocketIORoomStr,
+) -> dict[str, AsyncMock]:
+    # handlers
+    async def connect(sid: str, environ):
+        print("connecting", sid)
+        await socketio_server.enter_room(sid, room_name)
+
+    async def on_check(sid, data):
+        print("check", sid, data)
+
+    async def disconnect(sid: str):
+        print("disconnecting", sid)
+        await socketio_server.leave_room(sid, room_name)
+
+    # spies
+    spy_connect = mocker.AsyncMock(wraps=connect)
+    socketio_server.on("connect", spy_connect)
+
+    spy_on_check = mocker.AsyncMock(wraps=on_check)
+    socketio_server.on("check", spy_on_check)
+
+    spy_disconnect = mocker.AsyncMock(wraps=disconnect)
+    socketio_server.on("disconnect", spy_disconnect)
+
+    return {
+        connect.__name__: spy_connect,
+        disconnect.__name__: spy_disconnect,
+        on_check.__name__: spy_on_check,
+    }