fixed volume removal

Author: Andrei Neagu

services/agent/src/simcore_service_agent/core/api/rpc/_volumes.py

@@ -1,18 +1,25 @@
+import logging
+
 from fastapi import FastAPI
 from models_library.projects_nodes_io import NodeID
+from servicelib.logging_utils import log_context
 from servicelib.rabbitmq import RPCRouter
 from simcore_service_agent.services.volumes_manager import get_volumes_manager
 
+_logger = logging.getLogger(__name__)
+
 router = RPCRouter()
 
 
 @router.expose()
 async def remove_volumes_without_backup_for_service(
     app: FastAPI, *, node_id: NodeID
 ) -> None:
-    await get_volumes_manager(app).remove_service_volumes(node_id)
+    with log_context(_logger, logging.INFO, f"removing volumes for service: {node_id}"):
+        await get_volumes_manager(app).remove_service_volumes(node_id)
 
 
 @router.expose()
 async def backup_and_remove_volumes_for_all_services(app: FastAPI) -> None:
-    await get_volumes_manager(app).remove_all_volumes()
+    with log_context(_logger, logging.INFO, "removing all service volumes from node"):
+        await get_volumes_manager(app).remove_all_volumes()

services/agent/src/simcore_service_agent/core/api/rpc/routes.py

@@ -15,14 +15,14 @@ def setup_rpc_api_routes(app: FastAPI) -> None:
     async def startup() -> None:
         rpc_server = get_rabbitmq_rpc_server(app)
         settings: ApplicationSettings = app.state.settings
-        namespace = RPCNamespace.from_entries(
+        rpc_namespace = RPCNamespace.from_entries(
             {
                 "service": "agent",
                 "docker_node_id": settings.AGENT_DOCKER_NODE_ID,
                 "swarm_stack_name": settings.AGENT_VOLUMES_CLEANUP_TARGET_SWARM_STACK_NAME,
             }
         )
         for router in ROUTERS:
-            await rpc_server.register_router(router, namespace, app)
+            await rpc_server.register_router(router, rpc_namespace, app)
 
     app.add_event_handler("startup", startup)

services/agent/src/simcore_service_agent/services/volumes_manager.py

@@ -10,6 +10,13 @@
 from pydantic import NonNegativeFloat
 from servicelib.background_task import start_periodic_task, stop_periodic_task
 from servicelib.logging_utils import log_context
+from tenacity import (
+    AsyncRetrying,
+    TryAgain,
+    before_sleep_log,
+    stop_after_delay,
+    wait_fixed,
+)
 
 from ..core.settings import ApplicationSettings
 from .docker_utils import get_unused_dynamc_sidecar_volumes, remove_volume
@@ -96,17 +103,46 @@ async def _periodic_volmue_cleanup_task(self) -> None:
             for volume in volumes_to_remove:
                 await self._remove_volume_safe(volume_name=volume, requires_backup=True)
 
+    async def _wait_for_service_volumes_to_become_unused(
+        self, node_id: NodeID
+    ) -> set[str]:
+        # NOTE: it usually takes a few seconds for volumes to become unused
+        # if agent does not wait for this operation to finish
+        # they will be removed and backed up by the background task
+        # causing unncecessary data transfer to S3
+        async for attempt in AsyncRetrying(
+            reraise=True,
+            stop=stop_after_delay(60),
+            wait=wait_fixed(1),
+            before_sleep=before_sleep_log(_logger, logging.DEBUG),
+        ):
+            with attempt:
+                current_unused_volumes = await get_unused_dynamc_sidecar_volumes(
+                    self.docker
+                )
+
+                service_volumes = {
+                    v for v in current_unused_volumes if f"{node_id}" in v
+                }
+                _logger.debug(
+                    "service %s found volumes to remove: %s", node_id, service_volumes
+                )
+                if len(service_volumes) == 0:
+                    raise TryAgain
+
+        return current_unused_volumes
+
     async def remove_service_volumes(self, node_id: NodeID) -> None:
         # bookkept volumes might not be up to date
-        current_unused_volumes = await get_unused_dynamc_sidecar_volumes(self.docker)
-
-        service_volumes: set[str] = set()
-        for volume in current_unused_volumes:
-            if f"{node_id}" in volume:
-                service_volumes.add(volume)
+        service_volumes = await self._wait_for_service_volumes_to_become_unused(node_id)
+        _logger.debug(
+            "will remove volumes for %s from service_volumes=%s",
+            node_id,
+            service_volumes,
+        )
 
         for volume_name in service_volumes:
-            # these volumes have already been saved to S3 by the sidecar, no longer requires a backup
+            # volumes already saved to S3 by the sidecar and no longer require backup
             await self._remove_volume_safe(
                 volume_name=volume_name, requires_backup=False
             )

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

@@ -27,6 +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.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.utils import limited_gather, logged_gather
 from simcore_postgres_database.models.comp_tasks import NodeClass
 from tenacity import RetryError, TryAgain
@@ -230,9 +231,9 @@ async def service_remove_sidecar_proxy_docker_networks_and_volumes(
                 message="removing volumes", percent=ProgressPercent(0.3)
             )
             with log_context(_logger, logging.DEBUG, f"removing volumes '{node_uuid}'"):
-                rabbitmq_client: RabbitMQClient = app.state.rabbitmq_client
+                rabbit_rpc_client: RabbitMQRPCClient = app.state.rabbitmq_rpc_client
                 await remove_volumes_from_node(
-                    rabbitmq_client=rabbitmq_client,
+                    rabbit_rpc_client=rabbit_rpc_client,
                     docker_node_id=scheduler_data.dynamic_sidecar.docker_node_id,
                     swarm_stack_name=swarm_stack_name,
                     node_id=scheduler_data.node_uuid,

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

@@ -1,12 +1,12 @@
 from models_library.projects_nodes_io import NodeID
-from servicelib.rabbitmq import RabbitMQClient
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.agent.volumes import (
     remove_volumes_without_backup_for_service,
 )
 
 
 async def remove_volumes_from_node(
-    rabbitmq_client: RabbitMQClient,
+    rabbit_rpc_client: RabbitMQRPCClient,
     docker_node_id: str,
     swarm_stack_name: str,
     *,
@@ -15,7 +15,7 @@ async def remove_volumes_from_node(
     """removes all service volumes form the node where it was running"""
 
     await remove_volumes_without_backup_for_service(
-        rabbitmq_client,
+        rabbit_rpc_client,
         docker_node_id=docker_node_id,
         swarm_stack_name=swarm_stack_name,
         node_id=node_id,

services/docker-compose.yml

@@ -1010,6 +1010,7 @@ services:
       RABBIT_PASSWORD: ${RABBIT_PASSWORD}
       RABBIT_PORT: ${RABBIT_PORT}
       RABBIT_USER: ${RABBIT_USER}
+      RABBIT_SECURE: ${RABBIT_SECURE}
 
   dask-sidecar:
     image: ${DOCKER_REGISTRY:-itisfoundation}/dask-sidecar:${DOCKER_IMAGE_TAG:-latest}