🎨Dynamic-sidecar: report progress on sidecar task status (#8687)

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_api/_core.py

@@ -1,6 +1,6 @@
 import logging
 import re
-from collections.abc import Mapping
+from collections.abc import Awaitable, Callable, Mapping
 from typing import Any, Final
 
 import aiodocker
@@ -32,7 +32,12 @@
 )
 from ....models.dynamic_services_scheduler import NetworkId, SchedulerData, ServiceId
 from ....utils.dict_utils import get_leaf_key_paths, nested_update
-from ..docker_states import TASK_STATES_RUNNING, extract_task_state
+from ..docker_states import (
+    TASK_STATES_COMPLETE,
+    TASK_STATES_FAILED,
+    TASK_STATES_RUNNING,
+    extract_task_state,
+)
 from ..errors import DockerServiceNotFoundError, DynamicSidecarError, GenericDockerError
 from ._utils import docker_client
 
@@ -176,9 +181,24 @@ async def _get_service_latest_task(service_id: str) -> Mapping[str, Any]:
         raise
 
 
+_TASK_STATE_TO_PROGRESS_MAPPING: Final[dict[str, float]] = (
+    {
+        "pending": 0.2,
+        "assigned": 0.4,
+        "preparing": 0.6,
+        "starting": 0.8,
+        "running": 1.0,
+    }
+    | dict.fromkeys(TASK_STATES_COMPLETE, 1.0)
+    | dict.fromkeys(TASK_STATES_FAILED, 1.0)
+)
+
+
 async def get_dynamic_sidecar_placement(
     service_id: str,
     dynamic_services_scheduler_settings: DynamicServicesSchedulerSettings,
+    *,
+    progress_update: Callable[[float], Awaitable[None]] | None = None,
 ) -> DockerNodeID:
     """
     Waits until the service has a task in `running` state and
@@ -207,6 +227,12 @@ async def _get_task_data_when_service_running(service_id: str) -> Mapping[str, A
         """
         task = await _get_service_latest_task(service_id)
         service_state = task["Status"]["State"]
+        if progress_update:
+            # estimate progress based on task state
+
+            await progress_update(
+                _TASK_STATE_TO_PROGRESS_MAPPING.get(service_state, 0.0)
+            )
 
         if service_state not in TASK_STATES_RUNNING:
             raise TryAgain

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/_core/_event_create_sidecars.py

@@ -275,33 +275,50 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         dynamic_sidecar_service_final_spec = _merge_service_base_and_user_specs(
             dynamic_sidecar_service_spec_base, user_specific_service_spec
         )
-        rabbit_message = ProgressRabbitMessageNode.model_construct(
+        sidecar_pull_started_msg = ProgressRabbitMessageNode.model_construct(
             user_id=scheduler_data.user_id,
             project_id=scheduler_data.project_id,
             node_id=scheduler_data.node_uuid,
             progress_type=ProgressType.SIDECARS_PULLING,
             report=ProgressReport(actual_value=0, total=1),
         )
-        await rabbitmq_client.publish(rabbit_message.channel_name, rabbit_message)
+        await rabbitmq_client.publish(
+            ProgressRabbitMessageNode.get_channel_name(), sidecar_pull_started_msg
+        )
         dynamic_sidecar_id = await create_service_and_get_id(
             dynamic_sidecar_service_final_spec,
             app_settings.DIRECTOR_V2_DOCKER_HUB_REGISTRY,
         )
+        await rabbitmq_client.publish(
+            ProgressRabbitMessageNode.get_channel_name(),
+            sidecar_pull_started_msg.model_copy(
+                update={"report": ProgressReport(actual_value=0.1, total=1)}
+            ),
+        )
+
         # constrain service to the same node
+        async def progress_update(current: float) -> None:
+            await rabbitmq_client.publish(
+                ProgressRabbitMessageNode.get_channel_name(),
+                sidecar_pull_started_msg.model_copy(
+                    update={"report": ProgressReport(actual_value=current, total=1)}
+                ),
+            )
+
         scheduler_data.dynamic_sidecar.docker_node_id = (
             await get_dynamic_sidecar_placement(
-                dynamic_sidecar_id, dynamic_services_scheduler_settings
+                dynamic_sidecar_id,
+                dynamic_services_scheduler_settings,
+                progress_update=progress_update,
             )
         )
 
-        rabbit_message = ProgressRabbitMessageNode.model_construct(
-            user_id=scheduler_data.user_id,
-            project_id=scheduler_data.project_id,
-            node_id=scheduler_data.node_uuid,
-            progress_type=ProgressType.SIDECARS_PULLING,
-            report=ProgressReport(actual_value=1, total=1),
+        await rabbitmq_client.publish(
+            ProgressRabbitMessageNode.get_channel_name(),
+            sidecar_pull_started_msg.model_copy(
+                update={"report": ProgressReport(actual_value=1, total=1)}
+            ),
         )
-        await rabbitmq_client.publish(rabbit_message.channel_name, rabbit_message)
 
         await constrain_service_to_node(
             service_name=scheduler_data.service_name,