🎨 dynamic-sidecar pulls user services images with states and outputs (#6301)

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

Author: GitHK

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

@@ -111,6 +111,11 @@ class DynamicServicesSchedulerSettings(BaseCustomSettings):
         ),
     )
 
+    DYNAMIC_SIDECAR_API_USER_SERVICES_PULLING_TIMEOUT: PositiveFloat = Field(
+        60.0 * _MINUTE,
+        description="before starting the user services pull all the images in parallel",
+    )
+
     DYNAMIC_SIDECAR_API_RESTART_CONTAINERS_TIMEOUT: PositiveFloat = Field(
         1.0 * _MINUTE,
         description=(

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_public.py

@@ -280,6 +280,15 @@ async def detach_service_containers_from_project_network(
             ]
         )
 
+    async def submit_docker_compose_spec(
+        self,
+        dynamic_sidecar_endpoint: AnyHttpUrl,
+        compose_spec: str,
+    ) -> None:
+        await self._thin_client.post_containers_compose_spec(
+            dynamic_sidecar_endpoint, compose_spec=compose_spec
+        )
+
     def _get_client(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> Client:
         return Client(
             app=self._app,
@@ -307,13 +316,11 @@ async def _await_for_result(
     async def create_containers(
         self,
         dynamic_sidecar_endpoint: AnyHttpUrl,
-        compose_spec: str,
         metrics_params: CreateServiceMetricsAdditionalParams,
         progress_callback: ProgressCallback | None = None,
     ) -> None:
         response = await self._thin_client.post_containers_tasks(
             dynamic_sidecar_endpoint,
-            compose_spec=compose_spec,
             metrics_params=metrics_params,
         )
         task_id: TaskId = response.json()
@@ -355,6 +362,21 @@ async def restore_service_state(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> N
             _debug_progress_callback,
         )
 
+    async def pull_user_services_images(
+        self, dynamic_sidecar_endpoint: AnyHttpUrl
+    ) -> None:
+        response = await self._thin_client.post_containers_images_pull(
+            dynamic_sidecar_endpoint
+        )
+        task_id: TaskId = response.json()
+
+        await self._await_for_result(
+            task_id,
+            dynamic_sidecar_endpoint,
+            self._dynamic_services_scheduler_settings.DYNAMIC_SIDECAR_API_USER_SERVICES_PULLING_TIMEOUT,
+            _debug_progress_callback,
+        )
+
     async def save_service_state(
         self,
         dynamic_sidecar_endpoint: AnyHttpUrl,

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_thin.py

@@ -18,7 +18,7 @@
 )
 
 
-class ThinSidecarsClient(BaseThinClient):
+class ThinSidecarsClient(BaseThinClient):  # pylint: disable=too-many-public-methods
     """
     NOTE: all calls can raise the following errors.
     - `UnexpectedStatusError`
@@ -168,21 +168,26 @@ async def post_containers_networks_detach(
 
     @retry_on_errors()
     @expect_status(status.HTTP_202_ACCEPTED)
-    async def post_containers_tasks(
+    async def post_containers_compose_spec(
         self,
         dynamic_sidecar_endpoint: AnyHttpUrl,
         *,
         compose_spec: str,
+    ) -> Response:
+        url = self._get_url(dynamic_sidecar_endpoint, "/containers/compose-spec")
+        return await self.client.post(url, json={"docker_compose_yaml": compose_spec})
+
+    @retry_on_errors()
+    @expect_status(status.HTTP_202_ACCEPTED)
+    async def post_containers_tasks(
+        self,
+        dynamic_sidecar_endpoint: AnyHttpUrl,
+        *,
         metrics_params: CreateServiceMetricsAdditionalParams,
     ) -> Response:
         url = self._get_url(dynamic_sidecar_endpoint, "/containers")
-        # change introduce in OAS version==1.1.0
         return await self.client.post(
-            url,
-            json={
-                "docker_compose_yaml": compose_spec,
-                "metrics_params": metrics_params.dict(),
-            },
+            url, json={"metrics_params": metrics_params.dict()}
         )
 
     @retry_on_errors()
@@ -209,6 +214,14 @@ async def post_containers_tasks_state_save(
         url = self._get_url(dynamic_sidecar_endpoint, "/containers/state:save")
         return await self.client.post(url)
 
+    @retry_on_errors()
+    @expect_status(status.HTTP_202_ACCEPTED)
+    async def post_containers_images_pull(
+        self, dynamic_sidecar_endpoint: AnyHttpUrl
+    ) -> Response:
+        url = self._get_url(dynamic_sidecar_endpoint, "/containers/images:pull")
+        return await self.client.post(url)
+
     @retry_on_errors()
     @expect_status(status.HTTP_202_ACCEPTED)
     async def post_containers_tasks_ports_inputs_pull(

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

@@ -19,7 +19,7 @@
 from ...errors import UnexpectedContainerStatusError
 from ._abc import DynamicSchedulerEvent
 from ._event_create_sidecars import CreateSidecars
-from ._events_user_services import create_user_services
+from ._events_user_services import create_user_services, submit_compose_sepc
 from ._events_utils import (
     are_all_user_services_containers_running,
     attach_project_networks,
@@ -139,11 +139,33 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
             )
 
 
-class PrepareServicesEnvironment(DynamicSchedulerEvent):
+class SendUserServicesSpec(DynamicSchedulerEvent):
     """
     Triggered when the dynamic-sidecar is responding to http requests.
     This step runs before CreateUserServices.
 
+    Sends over the configuration that is used for all docker compose commands.
+    """
+
+    @classmethod
+    async def will_trigger(cls, app: FastAPI, scheduler_data: SchedulerData) -> bool:
+        assert app  # nose
+        return (
+            scheduler_data.dynamic_sidecar.status.current == DynamicSidecarStatus.OK
+            and scheduler_data.dynamic_sidecar.is_ready
+            and not scheduler_data.dynamic_sidecar.was_compose_spec_submitted
+        )
+
+    @classmethod
+    async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
+        await submit_compose_sepc(app, scheduler_data)
+
+
+class PrepareServicesEnvironment(DynamicSchedulerEvent):
+    """
+    Triggered when the dynamic-sidecar has it's docker-copose spec loaded.
+    This step runs before SendUserServicesSpec.
+
     Sets up the environment on the host required by the service.
     - restores service state
     """
@@ -174,7 +196,8 @@ async def will_trigger(cls, app: FastAPI, scheduler_data: SchedulerData) -> bool
         assert app  # nose
         return (
             scheduler_data.dynamic_sidecar.is_service_environment_ready
-            and not scheduler_data.dynamic_sidecar.compose_spec_submitted
+            and not scheduler_data.dynamic_sidecar.were_containers_created
+            and scheduler_data.dynamic_sidecar.compose_spec_submitted
         )
 
     @classmethod
@@ -236,6 +259,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
     WaitForSidecarAPI,
     UpdateHealth,
     GetStatus,
+    SendUserServicesSpec,
     PrepareServicesEnvironment,
     CreateUserServices,
     AttachProjectsNetworks,

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

@@ -32,9 +32,7 @@
 _logger = logging.getLogger(__name__)
 
 
-async def create_user_services(  # pylint: disable=too-many-statements
-    app: FastAPI, scheduler_data: SchedulerData
-) -> None:
+async def submit_compose_sepc(app: FastAPI, scheduler_data: SchedulerData) -> None:
     _logger.debug(
         "Getting docker compose spec for service %s", scheduler_data.service_name
     )
@@ -105,10 +103,26 @@ async def create_user_services(  # pylint: disable=too-many-statements
     )
 
     _logger.debug(
-        "Starting containers %s with compose-specs:\n%s",
+        "Submitting to %s it's compose-specs:\n%s",
         scheduler_data.service_name,
         compose_spec,
     )
+    await sidecars_client.submit_docker_compose_spec(
+        dynamic_sidecar_endpoint, compose_spec=compose_spec
+    )
+    scheduler_data.dynamic_sidecar.was_compose_spec_submitted = True
+
+
+async def create_user_services(  # pylint: disable=too-many-statements
+    app: FastAPI, scheduler_data: SchedulerData
+) -> None:
+    dynamic_services_scheduler_settings: DynamicServicesSchedulerSettings = (
+        app.state.settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER
+    )
+    sidecars_client = await get_sidecars_client(app, scheduler_data.node_uuid)
+    dynamic_sidecar_endpoint = scheduler_data.endpoint
+
+    _logger.debug("Starting containers %s", scheduler_data.service_name)
 
     async def progress_create_containers(
         message: str, percent: ProgressPercent | None, task_id: TaskId
@@ -159,7 +173,6 @@ async def progress_create_containers(
     )
     await sidecars_client.create_containers(
         dynamic_sidecar_endpoint,
-        compose_spec,
         metrics_params,
         progress_create_containers,
     )
@@ -209,6 +222,4 @@ async def progress_create_containers(
 
     scheduler_data.dynamic_sidecar.were_containers_created = True
 
-    scheduler_data.dynamic_sidecar.was_compose_spec_submitted = True
-
     _logger.info("Internal state after creating user services %s", scheduler_data)

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

@@ -27,7 +27,7 @@
 from servicelib.fastapi.long_running_tasks.server import TaskProgress
 from servicelib.logging_utils import log_context
 from servicelib.rabbitmq import RabbitMQClient
-from servicelib.utils import logged_gather
+from servicelib.utils import limited_gather, logged_gather
 from simcore_postgres_database.models.comp_tasks import NodeClass
 from tenacity import RetryError, TryAgain
 from tenacity.asyncio import AsyncRetrying
@@ -460,13 +460,16 @@ async def prepare_services_environment(
         )
     )
 
-    tasks = [sidecars_client.pull_service_output_ports(dynamic_sidecar_endpoint)]
+    tasks = [
+        sidecars_client.pull_user_services_images(dynamic_sidecar_endpoint),
+        sidecars_client.pull_service_output_ports(dynamic_sidecar_endpoint),
+    ]
     # When enabled no longer downloads state via nodeports
     # S3 is used to store state paths
     if not app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
         tasks.append(sidecars_client.restore_service_state(dynamic_sidecar_endpoint))
 
-    await logged_gather(*tasks, max_concurrency=2)
+    await limited_gather(*tasks, limit=3)
 
     # inside this directory create the missing dirs, fetch those form the labels
     director_v0_client: DirectorV0Client = get_director_v0_client(app)

services/director-v2/tests/unit/test_modules_dynamic_sidecar_client_api_public.py

@@ -166,7 +166,7 @@ async def test_is_healthy_api_error(
         "get_health",
         side_effect=side_effect,
     ) as client:
-        assert await client.is_healthy(dynamic_sidecar_endpoint) == False
+        assert await client.is_healthy(dynamic_sidecar_endpoint) is False
 
 
 async def test_containers_inspect(
@@ -282,11 +282,10 @@ async def test_get_entrypoint_container_name_api_not_found(
             ),
             expecting=status.HTTP_204_NO_CONTENT,
         ),
-    ) as client:
-        with pytest.raises(EntrypointContainerNotFoundError):
-            await client.get_entrypoint_container_name(
-                dynamic_sidecar_endpoint, dynamic_sidecar_network_name
-            )
+    ) as client, pytest.raises(EntrypointContainerNotFoundError):
+        await client.get_entrypoint_container_name(
+            dynamic_sidecar_endpoint, dynamic_sidecar_network_name
+        )
 
 
 @pytest.mark.parametrize("network_aliases", [[], ["an-alias"], ["alias-1", "alias-2"]])
@@ -301,7 +300,7 @@ async def test_attach_container_to_network(
     ) as client:
         assert (
             # pylint:disable=protected-access
-            await client._attach_container_to_network(
+            await client._attach_container_to_network(  # noqa: SLF001
                 dynamic_sidecar_endpoint,
                 container_id="container_id",
                 network_id="network_id",
@@ -321,7 +320,7 @@ async def test_detach_container_from_network(
     ) as client:
         assert (
             # pylint:disable=protected-access
-            await client._detach_container_from_network(
+            await client._detach_container_from_network(  # noqa: SLF001
                 dynamic_sidecar_endpoint,
                 container_id="container_id",
                 network_id="network_id",

services/director-v2/tests/unit/test_modules_dynamic_sidecar_client_api_thin.py

@@ -2,8 +2,8 @@
 # pylint:disable=redefined-outer-name
 
 import json
-from collections.abc import Callable
-from typing import Any, AsyncIterable
+from collections.abc import AsyncIterable, Callable
+from typing import Any
 
 import pytest
 from faker import Faker
@@ -282,11 +282,10 @@ async def test_put_volumes(
             "post_containers_tasks",
             "/containers",
             {
-                "compose_spec": "some_fake_compose_as_str",
                 "metrics_params": parse_obj_as(
                     CreateServiceMetricsAdditionalParams,
                     CreateServiceMetricsAdditionalParams.Config.schema_extra["example"],
-                ),
+                )
             },
             id="post_containers_tasks",
         ),
@@ -296,6 +295,12 @@ async def test_put_volumes(
             {},
             id="down",
         ),
+        pytest.param(
+            "post_containers_images_pull",
+            "/containers/images:pull",
+            {},
+            id="user_servces_images_pull",
+        ),
         pytest.param(
             "post_containers_tasks_state_restore",
             "/containers/state:restore",
@@ -387,3 +392,22 @@ async def test_post_disk_reserved_free(
 
     response = await thin_client.post_disk_reserved_free(dynamic_sidecar_endpoint)
     assert_responses(mock_response, response)
+
+
+async def test_post_containers_compose_spec(
+    thin_client: ThinSidecarsClient,
+    dynamic_sidecar_endpoint: AnyHttpUrl,
+    mock_request: MockRequestType,
+):
+    mock_response = Response(status.HTTP_202_ACCEPTED)
+    mock_request(
+        "POST",
+        f"{dynamic_sidecar_endpoint}/{thin_client.API_VERSION}/containers/compose-spec",
+        mock_response,
+        None,
+    )
+
+    response = await thin_client.post_containers_compose_spec(
+        dynamic_sidecar_endpoint, compose_spec="some_fake_compose_as_str"
+    )
+    assert_responses(mock_response, response)

services/dynamic-sidecar/VERSION

@@ -1 +1 @@
-1.1.1
+1.2.0