added force container cleanup step when stopping services

Author: Andrei Neagu

packages/models-library/src/models_library/api_schemas_directorv2/services.py

@@ -103,3 +103,7 @@ class ServiceExtras(BaseModel):
 
 
 CHARS_IN_VOLUME_NAME_BEFORE_DIR_NAME: Final[NonNegativeInt] = 89
+
+
+DYNAMIC_SIDECAR_SERVICE_PREFIX: Final[str] = "dy-sidecar"
+DYNAMIC_PROXY_SERVICE_PREFIX: Final[str] = "dy-proxy"

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/agent/containers.py

@@ -0,0 +1,36 @@
+import logging
+from datetime import timedelta
+from typing import Final
+
+from models_library.projects_nodes_io import NodeID
+from models_library.rabbitmq_basic_types import RPCMethodName, RPCNamespace
+from pydantic import NonNegativeInt, TypeAdapter
+from servicelib.logging_utils import log_decorator
+from servicelib.rabbitmq import RabbitMQRPCClient
+
+_logger = logging.getLogger(__name__)
+
+_REQUEST_TIMEOUT: Final[NonNegativeInt] = int(timedelta(minutes=60).total_seconds())
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def force_container_cleanup(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    docker_node_id: str,
+    swarm_stack_name: str,
+    node_id: NodeID,
+) -> None:
+    result = await rabbitmq_rpc_client.request(
+        RPCNamespace.from_entries(
+            {
+                "service": "agent",
+                "docker_node_id": docker_node_id,
+                "swarm_stack_name": swarm_stack_name,
+            }
+        ),
+        TypeAdapter(RPCMethodName).validate_python("force_container_cleanup"),
+        node_id=node_id,
+        timeout_s=_REQUEST_TIMEOUT,
+    )
+    assert result is None  # nosec

services/agent/src/simcore_service_agent/api/rpc/_containers.py

@@ -0,0 +1,20 @@
+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 ...services.containers_manager import ContainersManager
+
+_logger = logging.getLogger(__name__)
+
+router = RPCRouter()
+
+
+@router.expose()
+async def force_container_cleanup(app: FastAPI, *, node_id: NodeID) -> None:
+    with log_context(
+        _logger, logging.INFO, f"removing all orphan container for {node_id=}"
+    ):
+        await ContainersManager.get_from_app_state(app).force_container_cleanup(node_id)

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

@@ -7,7 +7,8 @@
 from servicelib.rabbitmq.rpc_interfaces.agent.errors import (
     NoServiceVolumesFoundRPCError,
 )
-from simcore_service_agent.services.volumes_manager import VolumesManager
+
+from ...services.volumes_manager import VolumesManager
 
 _logger = logging.getLogger(__name__)
 

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

@@ -4,9 +4,10 @@
 from simcore_service_agent.core.settings import ApplicationSettings
 
 from ...services.rabbitmq import get_rabbitmq_rpc_server
-from . import _volumes
+from . import _containers, _volumes
 
 ROUTERS: list[RPCRouter] = [
+    _containers.router,
     _volumes.router,
 ]
 

services/agent/src/simcore_service_agent/core/application.py

@@ -18,6 +18,7 @@
 )
 from ..api.rest.routes import setup_rest_api
 from ..api.rpc.routes import setup_rpc_api_routes
+from ..services.containers_manager import setup_containers_manager
 from ..services.instrumentation import setup_instrumentation
 from ..services.rabbitmq import setup_rabbitmq
 from ..services.volumes_manager import setup_volume_manager
@@ -58,6 +59,7 @@ def create_app() -> FastAPI:
 
     setup_rabbitmq(app)
     setup_volume_manager(app)
+    setup_containers_manager(app)
     setup_rest_api(app)
     setup_rpc_api_routes(app)
 

services/agent/src/simcore_service_agent/services/containers_manager.py

@@ -0,0 +1,62 @@
+import logging
+from dataclasses import dataclass, field
+
+from aiodocker import Docker
+from fastapi import FastAPI
+from models_library.api_schemas_directorv2.services import (
+    DYNAMIC_PROXY_SERVICE_PREFIX,
+    DYNAMIC_SIDECAR_SERVICE_PREFIX,
+)
+from models_library.projects_nodes_io import NodeID
+from servicelib.fastapi.app_state import SingletonInAppStateMixin
+from servicelib.utils import limited_gather
+
+from .docker_utils import get_containers_with_prefixes, remove_container_forcefully
+
+_logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ContainersManager(SingletonInAppStateMixin):
+    app_state_name: str = "containers_manager"
+
+    docker: Docker = field(default_factory=Docker)
+
+    async def force_container_cleanup(self, node_id: NodeID) -> None:
+        # compose all possible used container prefixes
+        proxy_prefix = f"{DYNAMIC_PROXY_SERVICE_PREFIX}_{node_id}"
+        dy_sidecar_prefix = f"{DYNAMIC_SIDECAR_SERVICE_PREFIX}_{node_id}"
+        user_service_prefix = f"{DYNAMIC_SIDECAR_SERVICE_PREFIX}-{node_id}"
+
+        orphan_containers = await get_containers_with_prefixes(
+            self.docker, {proxy_prefix, dy_sidecar_prefix, user_service_prefix}
+        )
+
+        _logger.debug(
+            "Orphan containers for node_id='%s': %s", node_id, orphan_containers
+        )
+
+        await limited_gather(
+            *[
+                remove_container_forcefully(self.docker, container)
+                for container in orphan_containers
+            ],
+        )
+
+    async def shutdown(self) -> None:
+        await self.docker.close()
+
+
+def get_containers_manager(app: FastAPI) -> ContainersManager:
+    return ContainersManager.get_from_app_state(app)
+
+
+def setup_containers_manager(app: FastAPI) -> None:
+    async def _on_startup() -> None:
+        ContainersManager().set_to_app_state(app)
+
+    async def _on_shutdown() -> None:
+        await ContainersManager.get_from_app_state(app).shutdown()
+
+    app.add_event_handler("startup", _on_startup)
+    app.add_event_handler("shutdown", _on_shutdown)

services/agent/src/simcore_service_agent/services/docker_utils.py

@@ -106,3 +106,27 @@ async def remove_volume(
         get_instrumentation(app).agent_metrics.remove_volumes(
             settings.AGENT_DOCKER_NODE_ID
         )
+
+
+async def get_containers_with_prefixes(docker: Docker, prefixes: set[str]) -> set[str]:
+    """Returns a set of container names matching any of the given prefixes"""
+    all_containers = await docker.containers.list(all=True)
+
+    result: set[str] = set()
+    for container in all_containers:
+        container_info = await container.show()
+        container_name = container_info.get("Name", "").lstrip("/")
+        if any(container_name.startswith(prefix) for prefix in prefixes):
+            result.add(container_name)
+
+    return result
+
+
+async def remove_container_forcefully(docker: Docker, container_id: str) -> None:
+    """Removes a container regardless of it's state"""
+    try:
+        container = await docker.containers.get(container_id)
+        await container.delete(force=True)
+    except DockerError as e:
+        if e.status != status.HTTP_404_NOT_FOUND:
+            raise

services/agent/tests/unit/test_api_rpc__containers.py

@@ -0,0 +1,54 @@
+# pylint:disable=redefined-outer-name
+# pylint:disable=unused-argument
+
+from collections.abc import Awaitable, Callable
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_mock
+from faker import Faker
+from fastapi import FastAPI
+from models_library.projects_nodes_io import NodeID
+from servicelib.rabbitmq import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.agent import containers
+
+pytest_simcore_core_services_selection = [
+    "rabbit",
+]
+
+
+@pytest.fixture
+def node_id(faker: Faker) -> NodeID:
+    return faker.uuid4(cast_to=None)
+
+
+@pytest.fixture
+async def rpc_client(
+    initialized_app: FastAPI,
+    rabbitmq_rpc_client: Callable[[str], Awaitable[RabbitMQRPCClient]],
+) -> RabbitMQRPCClient:
+    return await rabbitmq_rpc_client("client")
+
+
+@pytest.fixture
+def mocked_force_container_cleanup(mocker: pytest_mock.MockerFixture) -> AsyncMock:
+    return mocker.patch(
+        "simcore_service_agent.services.containers_manager.ContainersManager.force_container_cleanup"
+    )
+
+
+async def test_force_container_cleanup(
+    rpc_client: RabbitMQRPCClient,
+    swarm_stack_name: str,
+    docker_node_id: str,
+    node_id: NodeID,
+    mocked_force_container_cleanup: AsyncMock,
+):
+    assert mocked_force_container_cleanup.call_count == 0
+    await containers.force_container_cleanup(
+        rpc_client,
+        docker_node_id=docker_node_id,
+        swarm_stack_name=swarm_stack_name,
+        node_id=node_id,
+    )
+    assert mocked_force_container_cleanup.call_count == 1

services/agent/tests/unit/test_services_containers_manager.py

@@ -0,0 +1,107 @@
+# pylint: disable=redefined-outer-name
+
+
+import logging
+from collections.abc import AsyncIterable, Awaitable, Callable
+from enum import Enum
+
+import pytest
+from aiodocker import Docker, DockerError
+from asgi_lifespan import LifespanManager
+from faker import Faker
+from fastapi import FastAPI, status
+from models_library.api_schemas_directorv2.services import (
+    DYNAMIC_PROXY_SERVICE_PREFIX,
+    DYNAMIC_SIDECAR_SERVICE_PREFIX,
+)
+from models_library.projects_nodes_io import NodeID
+from simcore_service_agent.services.containers_manager import (
+    get_containers_manager,
+    setup_containers_manager,
+)
+
+
+@pytest.fixture
+async def app() -> AsyncIterable[FastAPI]:
+    app = FastAPI()
+    setup_containers_manager(app)
+
+    async with LifespanManager(app):
+        yield app
+
+
+@pytest.fixture
+def node_id(faker: Faker) -> NodeID:
+    return faker.uuid4(cast_to=None)
+
+
+@pytest.fixture
+async def docker() -> AsyncIterable[Docker]:
+    async with Docker() as docker:
+        yield docker
+
+
+class _ContainerMode(Enum):
+    CREATED = "CREATED"
+    RUNNING = "RUNNING"
+    STOPPED = "STOPPED"
+
+
+@pytest.fixture
+async def create_container(
+    docker: Docker,
+) -> AsyncIterable[Callable[[str, _ContainerMode], Awaitable[str]]]:
+    created_containers: set[str] = set()
+
+    async def _(name: str, container_mode: _ContainerMode) -> str:
+        container = await docker.containers.create(
+            config={
+                "Image": "alpine",
+                "Cmd": ["sh", "-c", "while true; do sleep 1; done"],
+            },
+            name=name,
+        )
+
+        if container_mode in (_ContainerMode.RUNNING, _ContainerMode.STOPPED):
+            await container.start()
+        if container_mode == _ContainerMode.STOPPED:
+            await container.stop()
+
+        created_containers.add(container.id)
+        return container.id
+
+    yield _
+
+    # cleanup containers
+    for container_id in created_containers:
+        try:
+            container = await docker.containers.get(container_id)
+            await container.delete(force=True)
+        except DockerError as e:
+            if e.status != status.HTTP_404_NOT_FOUND:
+                raise
+
+
+async def test_force_container_cleanup(
+    app: FastAPI,
+    node_id: NodeID,
+    create_container: Callable[[str, _ContainerMode], Awaitable[str]],
+    faker: Faker,
+    caplog: pytest.LogCaptureFixture,
+):
+    caplog.set_level(logging.DEBUG)
+    caplog.clear()
+
+    proxy_name = f"{DYNAMIC_PROXY_SERVICE_PREFIX}_{node_id}{faker.pystr()}"
+    dynamic_sidecar_name = f"{DYNAMIC_SIDECAR_SERVICE_PREFIX}-{node_id}{faker.pystr()}"
+    user_service_name = f"{DYNAMIC_SIDECAR_SERVICE_PREFIX}_{node_id}{faker.pystr()}"
+
+    await create_container(proxy_name, _ContainerMode.CREATED)
+    await create_container(dynamic_sidecar_name, _ContainerMode.RUNNING)
+    await create_container(user_service_name, _ContainerMode.STOPPED)
+
+    await get_containers_manager(app).force_container_cleanup(node_id)
+
+    assert proxy_name in caplog.text
+    assert dynamic_sidecar_name in caplog.text
+    assert user_service_name in caplog.text