current WIP version with FastUI

Author: Andrei Neagu

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/ui/_services.py

@@ -10,20 +10,31 @@
 from fastui import AnyComponent, FastUI
 from fastui import components as c
 from fastui.events import GoToEvent, PageEvent
+from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
+    DynamicServiceStop,
+)
 from models_library.projects_nodes_io import NodeID
 from servicelib.background_task import start_periodic_task, stop_periodic_task
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.logging_utils import log_catch, log_context
+from servicelib.rabbitmq.rpc_interfaces.dynamic_scheduler.services import (
+    stop_dynamic_service,
+)
+from simcore_service_dynamic_scheduler.services.service_tracker._models import (
+    SchedulerServiceState,
+)
 from starlette import status
 
+from ...core.settings import ApplicationSettings
+from ...services.rabbitmq import get_rabbitmq_rpc_client
 from ...services.service_tracker import (
     TrackedServiceModel,
     get_all_tracked_services,
     get_tracked_service,
 )
 from ..dependencies import get_app
 from . import _custom_components as cu
-from ._constants import API_ROOT_PATH
+from ._constants import API_ROOT_PATH, UI_MOUNT_PREFIX
 from ._sse_utils import (
     AbstractSSERenderer,
     render_items_on_change,
@@ -51,10 +62,7 @@ def _page_base(
     ]
 
 
-@router.get(
-    f"{API_ROOT_PATH}/", response_model=FastUI, response_model_exclude_none=True
-)
-def api_index() -> list[AnyComponent]:
+def _get_index_page() -> list[AnyComponent]:
     return _page_base(
         c.Heading(text="Dynamic services status", level=4),
         c.Paragraph(text="List of all services currently tracked by the scheduler"),
@@ -72,6 +80,13 @@ def api_index() -> list[AnyComponent]:
     )
 
 
+@router.get(
+    f"{API_ROOT_PATH}/", response_model=FastUI, response_model_exclude_none=True
+)
+def api_index() -> list[AnyComponent]:
+    return _get_index_page()
+
+
 @router.get(
     f"{API_ROOT_PATH}{_PREFIX}/details/",
     response_model=FastUI,
@@ -96,42 +111,151 @@ async def service_details(
     )
 
 
+@router.post(
+    f"{API_ROOT_PATH}{_PREFIX}/stop-service/",
+    response_model=FastUI,
+    response_model_exclude_none=True,
+)
+async def stop_service(
+    node_id: NodeID, app: Annotated[FastAPI, Depends(get_app)]
+) -> list[AnyComponent]:
+    service_model = await get_tracked_service(app, node_id)
+
+    if service_model and service_model.user_id and service_model.project_id:
+        settings: ApplicationSettings = app.state.settings
+        await stop_dynamic_service(
+            get_rabbitmq_rpc_client(app),
+            dynamic_service_stop=DynamicServiceStop(
+                user_id=service_model.user_id,
+                project_id=service_model.project_id,
+                node_id=node_id,
+                simcore_user_agent="",
+                save_state=True,
+            ),
+            timeout_s=int(
+                settings.DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT.total_seconds()
+            ),
+        )
+
+    return _get_index_page()
+
+
+# TODO: add a toast for displaying the status of the stop operation
+# TODO: change logic since SSE is not working as expected. You can use it to detect changes in the model  -> then reload the page, which is useless
+# TODO: add an endpoint to remove from tracking without using anything else
+
+
+def _render_partial(
+    node_id: NodeID, service_model: TrackedServiceModel
+) -> AnyComponent:
+    list_display: list[tuple[Any, Any]] = [
+        ("NodeID", node_id),
+        ("Service state", service_model.current_state),
+        (
+            "Last state change",
+            arrow.get(service_model.last_state_change).isoformat(),
+        ),
+        ("Requested", service_model.requested_state),
+        ("ProjectID", service_model.project_id),
+        ("UserID", service_model.user_id),
+    ]
+
+    if service_model.dynamic_service_start:
+        list_display.extend(
+            [
+                ("Service Key", service_model.dynamic_service_start.key),
+                ("Service Version", service_model.dynamic_service_start.version),
+                ("Product", service_model.dynamic_service_start.product_name),
+            ]
+        )
+    components = [
+        c.Text(text="PARTIAL"),
+        cu.markdown_list_display(list_display),
+        c.Button(
+            text="Details",
+            named_style="secondary",
+            on_click=GoToEvent(url=f"{_PREFIX}/details/?node_id={node_id}"),
+        ),
+        c.Button(
+            text="Stop Service",
+            on_click=PageEvent(name="modal-prompt"),
+            class_name="+ ms-2",
+        ),
+    ]
+
+    return c.Div(components=components, class_name="border border-double")
+
+
+def _render_full(node_id: NodeID, service_model: TrackedServiceModel) -> AnyComponent:
+    list_display: list[tuple[Any, Any]] = [
+        ("NodeID", node_id),
+        ("Service state", service_model.current_state),
+        (
+            "Last state change",
+            arrow.get(service_model.last_state_change).isoformat(),
+        ),
+        ("Requested", service_model.requested_state),
+        ("ProjectID", service_model.project_id),
+        ("UserID", service_model.user_id),
+    ]
+
+    if service_model.dynamic_service_start:
+        list_display.extend(
+            [
+                ("Service Key", service_model.dynamic_service_start.key),
+                ("Service Version", service_model.dynamic_service_start.version),
+                ("Product", service_model.dynamic_service_start.product_name),
+            ]
+        )
+    components = [
+        c.Text(text="FULL_RENDERED"),
+        cu.markdown_list_display(list_display),
+        c.Button(
+            text="Details",
+            named_style="secondary",
+            on_click=GoToEvent(url=f"{_PREFIX}/details/?node_id={node_id}"),
+        ),
+        c.Button(
+            text="Stop Service",
+            on_click=PageEvent(name="modal-prompt"),
+            class_name="+ ms-2",
+        ),
+        c.Modal(
+            title="Stop Service",
+            body=[
+                c.Paragraph(text=f"Are you sure you want to stop {node_id}?"),
+                c.Form(
+                    form_fields=[],
+                    submit_url=f"{UI_MOUNT_PREFIX}{API_ROOT_PATH}{_PREFIX}/stop-service/?node_id={node_id}",
+                    loading=[c.Spinner(text="Stopping...")],
+                    footer=[],
+                    submit_trigger=PageEvent(name="modal-form-submit"),
+                ),
+            ],
+            footer=[
+                c.Button(
+                    text="Cancel",
+                    named_style="secondary",
+                    on_click=PageEvent(name="modal-prompt", clear=True),
+                ),
+                c.Button(text="Submit", on_click=PageEvent(name="modal-form-submit")),
+            ],
+            open_trigger=PageEvent(name="modal-prompt"),
+        ),
+    ]
+
+    return c.Div(components=components, class_name="border border-dotted")
+
+
 class ServicesSSERenderer(AbstractSSERenderer):
     @staticmethod
     def get_component(item: tuple[NodeID, TrackedServiceModel]) -> AnyComponent:
         node_id, service_model = item
 
-        list_display: list[tuple[Any, Any]] = [
-            ("NodeID", node_id),
-            ("Service state", service_model.current_state),
-            (
-                "Last state change",
-                arrow.get(service_model.last_state_change).isoformat(),
-            ),
-            ("Requested", service_model.requested_state),
-            ("ProjectID", service_model.project_id),
-            ("UserID", service_model.user_id),
-        ]
-
-        if service_model.dynamic_service_start:
-            list_display.extend(
-                [
-                    ("Service Key", service_model.dynamic_service_start.key),
-                    ("Service Version", service_model.dynamic_service_start.version),
-                    ("Product", service_model.dynamic_service_start.product_name),
-                ]
-            )
-        components = [
-            cu.markdown_list_display(list_display),
-            c.Link(
-                components=[c.Text(text="Details")],
-                on_click=GoToEvent(
-                    url=f"{_PREFIX}/details/?node_id={node_id}",
-                ),
-            ),
-        ]
+        if service_model.current_state == SchedulerServiceState.RUNNING:
+            return _render_full(node_id, service_model)
 
-        return c.Div(components=components, class_name="border border-blue-500 px-4")
+        return _render_partial(node_id, service_model)
 
 
 @router.get(f"{API_ROOT_PATH}{_PREFIX}/sse/")
@@ -173,7 +297,7 @@ def startup(self) -> None:
         self._task = start_periodic_task(
             self._task_service_state_retrieval,
             interval=self.poll_interval,
-            task_name="sse_periodic_status_poll",
+            task_name="sse_services_status_retrieval_from_redis",
         )
 
     async def shutdown(self) -> None:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/rabbitmq.py

@@ -21,13 +21,18 @@ async def _on_startup() -> None:
         app.state.rabbitmq_client = RabbitMQClient(
             client_name="dynamic_scheduler", settings=settings
         )
+        app.state.rabbitmq_rpc_client = await RabbitMQRPCClient.create(
+            client_name="dynamic_scheduler_rpc_client", settings=settings
+        )
         app.state.rabbitmq_rpc_server = await RabbitMQRPCClient.create(
             client_name="dynamic_scheduler_rpc_server", settings=settings
         )
 
     async def _on_shutdown() -> None:
         if app.state.rabbitmq_client:
             await app.state.rabbitmq_client.close()
+        if app.state.rabbitmq_rpc_client:
+            await app.state.rabbitmq_rpc_client.close()
         if app.state.rabbitmq_rpc_server:
             await app.state.rabbitmq_rpc_server.close()
 
@@ -40,6 +45,11 @@ def get_rabbitmq_client(app: FastAPI) -> RabbitMQClient:
     return cast(RabbitMQClient, app.state.rabbitmq_client)
 
 
+def get_rabbitmq_rpc_client(app: FastAPI) -> RabbitMQRPCClient:
+    assert app.state.rabbitmq_rpc_client
+    return cast(RabbitMQRPCClient, app.state.rabbitmq_rpc_client)
+
+
 def get_rabbitmq_rpc_server(app: FastAPI) -> RabbitMQRPCClient:
     assert app.state.rabbitmq_rpc_server  # nosec
     return cast(RabbitMQRPCClient, app.state.rabbitmq_rpc_server)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/__init__.py

@@ -4,7 +4,7 @@
     get_tracked_service,
     get_user_id_for_service,
     remove_tracked_service,
-    set_frontned_notified_for_service,
+    set_frontend_notified_for_service,
     set_if_status_changed_for_service,
     set_request_as_running,
     set_request_as_stopped,
@@ -21,7 +21,7 @@
     "get_user_id_for_service",
     "NORMAL_RATE_POLL_INTERVAL",
     "remove_tracked_service",
-    "set_frontned_notified_for_service",
+    "set_frontend_notified_for_service",
     "set_if_status_changed_for_service",
     "set_request_as_running",
     "set_request_as_stopped",

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/service_tracker/_api.py

@@ -184,7 +184,7 @@ async def should_notify_frontend_for_service(
     )
 
 
-async def set_frontned_notified_for_service(app: FastAPI, node_id: NodeID) -> None:
+async def set_frontend_notified_for_service(app: FastAPI, node_id: NodeID) -> None:
     tracker = get_tracker(app)
     model: TrackedServiceModel | None = await tracker.load(node_id)
     if model is None:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_deferred_get_status.py

@@ -76,7 +76,9 @@ async def on_result(
             )
             if user_id:
                 await notify_service_status_change(app, user_id, result)
-                await service_tracker.set_frontned_notified_for_service(app, node_id)
+                # TODO: also notify the UI renderer about the change
+                # Write this using pubsub and a queue. We render if stuff changes?
+                await service_tracker.set_frontend_notified_for_service(app, node_id)
             else:
                 _logger.info(
                     "Did not find a user for '%s', skipping status delivery of: %s",