♻️ refactor dy-sidecar volume searching (ITISFoundation#3110)

* removal of code which no effect in swarm mode
* dy-sidecar now uses run_id to searach for volumes

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

Author: GitHK

services/director-v2/src/simcore_service_director_v2/models/schemas/dynamic_services/scheduler.py

@@ -3,7 +3,7 @@
 from asyncio import Lock
 from enum import Enum
 from typing import Any, Dict, List, Mapping, Optional
-from uuid import UUID
+from uuid import UUID, uuid4
 
 from models_library.projects_nodes_io import NodeID
 from models_library.service_settings_labels import (
@@ -128,6 +128,16 @@ def mark_removed(self) -> None:
 
 
 class DynamicSidecar(BaseModel):
+    run_id: UUID = Field(
+        default_factory=uuid4,
+        description=(
+            "Used to discriminate between dynamic-sidecar docker resources "
+            "generated during different runs. Sometimes artifacts remain in the"
+            "system after an error. This helps avoiding collisions."
+            "For now used by anonymous volumes involved in data sharing"
+        ),
+    )
+
     status: Status = Field(
         Status.create_as_initially_ok(),
         description="status of the service sidecar also with additional information",

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

@@ -7,7 +7,7 @@
 import time
 import traceback
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Dict, List, Mapping, Optional, Set, Tuple
+from typing import Any, AsyncIterator, Mapping, Optional
 
 import aiodocker
 from aiodocker.utils import clean_filters, clean_map
@@ -17,7 +17,6 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
-from packaging import version
 from servicelib.utils import logged_gather
 from tenacity._asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
@@ -43,49 +42,6 @@
 log = logging.getLogger(__name__)
 
 
-def _monkey_patch_aiodocker() -> None:
-    """Raises an error once the library is up to date."""
-    from aiodocker import volumes
-    from aiodocker.volumes import DockerVolume
-
-    if version.parse(aiodocker.__version__) > version.parse("0.21.0"):
-        raise RuntimeError(
-            "Please check that PR https://github.com/aio-libs/aiodocker/pull/623 "
-            "is not part of the current bump version. "
-            "Otherwise, if the current PR is part of this new release "
-            "remove monkey_patch."
-        )
-
-    # pylint: disable=protected-access
-    async def _custom_volumes_list(self, *, filters=None):
-        """
-        Return a list of volumes
-
-        Args:
-            filters: a dict with a list of filters
-
-        Available filters:
-            dangling=<boolean>
-            driver=<volume-driver-name>
-            label=<key> or label=<key>:<value>
-            name=<volume-name>
-        """
-        params = {} if filters is None else {"filters": clean_filters(filters)}
-
-        data = await self.docker._query_json("volumes", params=params)
-        return data
-
-    async def _custom_volumes_get(self, id):  # pylint: disable=redefined-builtin
-        data = await self.docker._query_json("volumes/{id}".format(id=id), method="GET")
-        return DockerVolume(self.docker, data["Name"])
-
-    setattr(volumes.DockerVolumes, "list", _custom_volumes_list)
-    setattr(volumes.DockerVolumes, "get", _custom_volumes_get)
-
-
-_monkey_patch_aiodocker()
-
-
 class _RetryError(Exception):
     pass
 
@@ -106,7 +62,7 @@ async def docker_client() -> AsyncIterator[aiodocker.docker.Docker]:
             await client.close()
 
 
-async def get_swarm_network(dynamic_sidecar_settings: DynamicSidecarSettings) -> Dict:
+async def get_swarm_network(dynamic_sidecar_settings: DynamicSidecarSettings) -> dict:
     async with docker_client() as client:
         all_networks = await client.networks.list()
 
@@ -127,7 +83,7 @@ async def get_swarm_network(dynamic_sidecar_settings: DynamicSidecarSettings) ->
     return networks[0]
 
 
-async def create_network(network_config: Dict[str, Any]) -> str:
+async def create_network(network_config: dict[str, Any]) -> str:
     async with docker_client() as client:
         try:
             docker_network = await client.networks.create(network_config)
@@ -168,17 +124,17 @@ async def create_service_and_get_id(create_service_data: AioDockerServiceSpec) -
     return service_start_result["ID"]
 
 
-async def inspect_service(service_id: str) -> Dict[str, Any]:
+async def inspect_service(service_id: str) -> dict[str, Any]:
     async with docker_client() as client:
         return await client.services.inspect(service_id)
 
 
 async def get_dynamic_sidecars_to_observe(
     dynamic_sidecar_settings: DynamicSidecarSettings,
-) -> List[SchedulerData]:
+) -> list[SchedulerData]:
     """called when scheduler is started to discover new services to observe"""
     async with docker_client() as client:
-        running_dynamic_sidecar_services: List[
+        running_dynamic_sidecar_services: list[
             Mapping[str, Any]
         ] = await client.services.list(
             filters={
@@ -196,12 +152,12 @@ async def get_dynamic_sidecars_to_observe(
 async def _extract_task_data_from_service_for_state(
     service_id: str,
     dynamic_sidecar_settings: DynamicSidecarSettings,
-    target_statuses: Set[str],
-) -> Dict[str, Any]:
+    target_statuses: set[str],
+) -> dict[str, Any]:
     """Waits until the dynamic-sidecar task is in one of the target_statuses
     and then returns the task"""
 
-    async def _sleep_or_error(started: float, task: Dict):
+    async def _sleep_or_error(started: float, task: dict):
         await asyncio.sleep(1.0)
         elapsed = time.time() - started
         if (
@@ -217,7 +173,7 @@ async def _sleep_or_error(started: float, task: Dict):
 
     async with docker_client() as client:
         service_state: Optional[str] = None
-        task: Dict[str, Any] = {}
+        task: dict[str, Any] = {}
 
         started = time.time()
 
@@ -275,8 +231,8 @@ async def get_service_placement(
     return task["NodeID"]
 
 
-async def get_dynamic_sidecar_state(service_id: str) -> Tuple[ServiceState, str]:
-    def _make_pending() -> Tuple[ServiceState, str]:
+async def get_dynamic_sidecar_state(service_id: str) -> tuple[ServiceState, str]:
+    def _make_pending() -> tuple[ServiceState, str]:
         pending_task_state = {"State": ServiceState.PENDING.value}
         return extract_task_state(task_status=pending_task_state)
 
@@ -386,38 +342,11 @@ async def remove_dynamic_sidecar_network(network_name: str) -> bool:
         return False
 
 
-async def remove_dynamic_sidecar_volumes(
-    node_uuid: NodeID, dynamic_sidecar_settings: DynamicSidecarSettings
-) -> Set[str]:
-    async with docker_client() as client:
-        volumes_response = await client.volumes.list(
-            filters={
-                "label": [
-                    f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}",
-                    f"uuid={node_uuid}",
-                ]
-            }
-        )
-        volumes = volumes_response["Volumes"]
-        log.debug("Removing volumes: %s", [v["Name"] for v in volumes])
-        if len(volumes) == 0:
-            log.warning("Expected to find at least 1 volume to remove, 0 were found")
-
-        removed_volumes: Set[str] = set()
-
-        for volume_data in volumes:
-            volume = await client.volumes.get(volume_data["Name"])
-            await volume.delete()
-            removed_volumes.add(volume_data["Name"])
-
-        return removed_volumes
-
-
 async def list_dynamic_sidecar_services(
     dynamic_sidecar_settings: DynamicSidecarSettings,
     user_id: Optional[UserID] = None,
     project_id: Optional[ProjectID] = None,
-) -> List[Dict[str, Any]]:
+) -> list[dict[str, Any]]:
     service_filters = {
         "label": [
             f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}",
@@ -451,8 +380,8 @@ async def is_dynamic_service_running(
 
 
 async def get_or_create_networks_ids(
-    networks: List[str], project_id: ProjectID
-) -> Dict[str, str]:
+    networks: list[str], project_id: ProjectID
+) -> dict[str, str]:
     async def _get_id_from_name(client, network_name: str) -> str:
         network = await client.networks.get(network_name)
         network_inspect = await network.show()
@@ -496,7 +425,7 @@ async def _get_id_from_name(client, network_name: str) -> str:
 
 async def get_projects_networks_containers(
     project_id: ProjectID,
-) -> Dict[str, int]:
+) -> dict[str, int]:
     """
     Returns all current projects_networks for the project with
     the amount of containers attached to them.
@@ -511,8 +440,8 @@ async def get_projects_networks_containers(
     if not filtered_networks:
         return {}
 
-    def _count_containers(item: Dict[str, Any]) -> int:
-        containers: Optional[List] = item.get("Containers")
+    def _count_containers(item: dict[str, Any]) -> int:
+        containers: Optional[list] = item.get("Containers")
         return 0 if containers is None else len(containers)
 
     return {x["Name"]: _count_containers(x) for x in filtered_networks}

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/sidecar.py

@@ -49,6 +49,7 @@ def _get_environment_variables(
         "DY_SIDECAR_USER_ID": f"{scheduler_data.user_id}",
         "DY_SIDECAR_PROJECT_ID": f"{scheduler_data.project_id}",
         "DY_SIDECAR_NODE_ID": f"{scheduler_data.node_uuid}",
+        "DY_SIDECAR_RUN_ID": f"{scheduler_data.dynamic_sidecar.run_id}",
         "POSTGRES_HOST": f"{app_settings.POSTGRES.POSTGRES_HOST}",
         "POSTGRES_ENDPOINT": f"{app_settings.POSTGRES.POSTGRES_HOST}:{app_settings.POSTGRES.POSTGRES_PORT}",
         "POSTGRES_PASSWORD": f"{app_settings.POSTGRES.POSTGRES_PASSWORD.get_secret_value()}",
@@ -121,6 +122,7 @@ def get_dynamic_sidecar_spec(
                 compose_namespace=compose_namespace,
                 path=path_to_mount,
                 node_uuid=scheduler_data.node_uuid,
+                run_id=scheduler_data.dynamic_sidecar.run_id,
             )
         )
     # state paths now get mounted via different driver and are synced to s3 automatically
@@ -134,6 +136,7 @@ def get_dynamic_sidecar_spec(
                     path=path_to_mount,
                     project_id=scheduler_data.project_id,
                     node_uuid=scheduler_data.node_uuid,
+                    run_id=scheduler_data.dynamic_sidecar.run_id,
                     r_clone_settings=dynamic_sidecar_settings.DYNAMIC_SIDECAR_R_CLONE_SETTINGS,
                 )
             )
@@ -144,6 +147,7 @@ def get_dynamic_sidecar_spec(
                     compose_namespace=compose_namespace,
                     path=path_to_mount,
                     node_uuid=scheduler_data.node_uuid,
+                    run_id=scheduler_data.dynamic_sidecar.run_id,
                 )
             )
 

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

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Coroutine, Dict, Final, List, Optional, Set, Type, cast
+from typing import Any, Coroutine, Final, Optional, Type, cast
 
 import httpx
 from fastapi import FastAPI
@@ -20,15 +20,14 @@
 from tenacity._asyncio import AsyncRetrying
 from tenacity.before_sleep import before_sleep_log
 from tenacity.stop import stop_after_delay
-from tenacity.wait import wait_exponential, wait_fixed
+from tenacity.wait import wait_fixed
 
 from ....core.settings import AppSettings, DynamicSidecarSettings
 from ....models.schemas.dynamic_services import DynamicSidecarStatus, SchedulerData
 from ....modules.director_v0 import DirectorV0Client
 from ...catalog import CatalogClient
 from ...db.repositories.projects import ProjectsRepository
 from ...db.repositories.projects_networks import ProjectsNetworksRepository
-from .._namepsace import get_compose_namespace
 from ..client_api import DynamicSidecarClient, get_dynamic_sidecar_client
 from ..docker_api import (
     are_all_services_present,
@@ -41,7 +40,6 @@
     is_dynamic_sidecar_missing,
     remove_dynamic_sidecar_network,
     remove_dynamic_sidecar_stack,
-    remove_dynamic_sidecar_volumes,
     try_to_remove_network,
 )
 from ..docker_compose_specs import assemble_spec
@@ -54,10 +52,8 @@
 from ..errors import (
     DynamicSidecarUnexpectedResponseStatus,
     EntrypointContainerNotFoundError,
-    GenericDockerError,
     NodeportsDidNotFindNodeError,
 )
-from ..volumes_resolver import DynamicSidecarVolumesPathsResolver
 from .abc import DynamicSchedulerEvent
 from .events_utils import (
     all_containers_running,
@@ -145,7 +141,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         dynamic_sidecar_network_id = await create_network(network_config)
 
         # attach the service to the swarm network dedicated to services
-        swarm_network: Dict[str, Any] = await get_swarm_network(
+        swarm_network: dict[str, Any] = await get_swarm_network(
             dynamic_sidecar_settings
         )
         swarm_network_id: str = swarm_network["Id"]
@@ -231,7 +227,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         dynamic_sidecar_endpoint = scheduler_data.dynamic_sidecar.endpoint
 
         try:
-            containers_inspect: Dict[
+            containers_inspect: dict[
                 str, Any
             ] = await dynamic_sidecar_client.containers_inspect(
                 dynamic_sidecar_endpoint
@@ -586,49 +582,9 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
             scheduler_data.dynamic_sidecar_network_name
         )
 
-        # remove created inputs and outputs volumes
-
-        # compute which volumes we expected to be removed
-        # in case the expected volumes differ from the removed ones
-        # show an error
-        compose_namespace = get_compose_namespace(scheduler_data.node_uuid)
-        expected_volumes_to_remove: Set[str] = {
-            DynamicSidecarVolumesPathsResolver.source(
-                compose_namespace=compose_namespace, path=path
-            )
-            for path in [
-                scheduler_data.paths_mapping.inputs_path,
-                scheduler_data.paths_mapping.outputs_path,
-            ]
-            + scheduler_data.paths_mapping.state_paths
-        }
-
-        async for attempt in AsyncRetrying(
-            wait=wait_exponential(min=1),
-            stop=stop_after_delay(
-                dynamic_sidecar_settings.DYNAMIC_SIDECAR_VOLUMES_REMOVAL_TIMEOUT_S
-            ),
-            retry_error_cls=GenericDockerError,
-        ):
-            with attempt:
-                logger.info(
-                    "Trying to remove volumes for %s", scheduler_data.service_name
-                )
-
-                removed_volumes = await remove_dynamic_sidecar_volumes(
-                    scheduler_data.node_uuid, dynamic_sidecar_settings
-                )
-
-                if expected_volumes_to_remove != removed_volumes:
-                    logger.warning(
-                        (
-                            "Attention expected to remove %s, instead only removed %s. "
-                            "Please check with check that all expected to remove volumes "
-                            "are now gone."
-                        ),
-                        expected_volumes_to_remove,
-                        removed_volumes,
-                    )
+        # NOTE: for future attempts, volumes cannot be cleaned up
+        # since they are local to the node.
+        # That's why anonymous volumes are used!
 
         logger.debug(
             "Removed dynamic-sidecar created services for '%s'",
@@ -657,7 +613,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
 
 # register all handlers defined in this module here
 # A list is essential to guarantee execution order
-REGISTERED_EVENTS: List[Type[DynamicSchedulerEvent]] = [
+REGISTERED_EVENTS: list[Type[DynamicSchedulerEvent]] = [
     CreateSidecars,
     GetStatus,
     PrepareServicesEnvironment,