♻️ connecting docker-api-proxy to services which require docker swarm socket

packages/pytest-simcore/src/pytest_simcore/docker_api_proxy.py

@@ -1,8 +1,13 @@
 import logging
+from collections.abc import Callable
+from contextlib import AsyncExitStack
 
+import aiodocker
 import pytest
 from aiohttp import BasicAuth, ClientSession, ClientTimeout
+from fastapi import FastAPI
 from pydantic import TypeAdapter
+from pytest_mock.plugin import MockerFixture
 from settings_library.docker_api_proxy import DockerApiProxysettings
 from tenacity import before_sleep_log, retry, stop_after_delay, wait_fixed
 
@@ -58,3 +63,24 @@ async def docker_api_proxy_settings(
     await _wait_till_docker_api_proxy_is_responsive(settings)
 
     return settings
+
+
+@pytest.fixture
+async def mock_setup_remote_docker_client(mocker: MockerFixture) -> Callable[[str], None]:
+    def _(target_setip_to_replace: str) -> None:
+        def _setup(app: FastAPI, settings: DockerApiProxysettings) -> None:
+            _ = settings
+            exit_stack = AsyncExitStack()
+
+            async def on_startup() -> None:
+                app.state.remote_docker_client = await exit_stack.enter_async_context(aiodocker.Docker())
+
+            async def on_shutdown() -> None:
+                await exit_stack.aclose()
+
+            app.add_event_handler("startup", on_startup)
+            app.add_event_handler("shutdown", on_shutdown)
+
+        mocker.patch(target_setip_to_replace, new=_setup)
+
+    return _

packages/service-library/src/servicelib/fastapi/docker.py

@@ -48,6 +48,31 @@ async def remote_docker_client_lifespan(app: FastAPI, state: State) -> AsyncIter
         yield {}
 
 
+def setup_remote_docker_client(app: FastAPI, settings: DockerApiProxysettings) -> None:
+    exit_stack = AsyncExitStack()
+
+    async def on_startup() -> None:
+        session = await exit_stack.enter_async_context(
+            ClientSession(
+                auth=aiohttp.BasicAuth(
+                    login=settings.DOCKER_API_PROXY_USER,
+                    password=settings.DOCKER_API_PROXY_PASSWORD.get_secret_value(),
+                )
+            )
+        )
+
+        app.state.remote_docker_client = await exit_stack.enter_async_context(
+            aiodocker.Docker(url=settings.base_url, session=session)
+        )
+        await wait_till_docker_api_proxy_is_responsive(app)
+
+    async def on_shutdown() -> None:
+        await exit_stack.aclose()
+
+    app.add_event_handler("startup", on_startup)
+    app.add_event_handler("shutdown", on_shutdown)
+
+
 @tenacity.retry(
     wait=tenacity.wait_fixed(5),
     stop=tenacity.stop_after_delay(60),

services/autoscaling/src/simcore_service_autoscaling/core/application.py

@@ -2,6 +2,7 @@
 
 from common_library.json_serialization import json_dumps
 from fastapi import FastAPI
+from servicelib.fastapi.docker import setup_remote_docker_client
 from servicelib.fastapi.tracing import (
     initialize_fastapi_app_tracing,
     setup_tracing,
@@ -61,6 +62,10 @@ def create_app(settings: ApplicationSettings, tracing_config: TracingConfig) ->
 
     setup_instrumentation(app)
     setup_api_routes(app)
+
+    if settings.AUTOSCALING_DOCKER_API_PROXY:
+        setup_remote_docker_client(app, settings.AUTOSCALING_DOCKER_API_PROXY)
+
     setup_docker(app)
     setup_rabbitmq(app)
     setup_ec2(app)

services/autoscaling/src/simcore_service_autoscaling/core/settings.py

@@ -24,6 +24,7 @@
 from servicelib.logging_utils import LogLevelInt
 from settings_library.application import BaseApplicationSettings
 from settings_library.base import BaseCustomSettings
+from settings_library.docker_api_proxy import DockerApiProxysettings
 from settings_library.docker_registry import RegistrySettings
 from settings_library.ec2 import EC2Settings
 from settings_library.rabbit import RabbitSettings
@@ -61,8 +62,11 @@ class EC2InstancesSettings(BaseCustomSettings):
     EC2_INSTANCES_ALLOWED_TYPES: Annotated[
         Json[dict[str, EC2InstanceBootSpecific]],
         Field(
-            description="Defines which EC2 instances are considered as candidates for new EC2 instance and their respective boot specific parameters"
-            "NOTE: minimum length >0",
+            description=(
+                "Defines which EC2 instances are considered as candidates for new "
+                "EC2 instance and their respective boot specific parameters"
+                "NOTE: minimum length >0"
+            ),
         ),
     ]
 
@@ -100,9 +104,11 @@ class EC2InstancesSettings(BaseCustomSettings):
         datetime.timedelta,
         Field(
             description="Usual time taken an EC2 instance with the given AMI takes to join the cluster "
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)."
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types "
+            "for string formatting)."
             "NOTE: be careful that this time should always be a factor larger than the real time, as EC2 instances"
-            "that take longer than this time will be terminated as sometimes it happens that EC2 machine fail on start.",
+            "that take longer than this time will be terminated as sometimes it happens that EC2 machine "
+            "fail on start.",
         ),
     ] = datetime.timedelta(minutes=1)
 
@@ -118,7 +124,8 @@ class EC2InstancesSettings(BaseCustomSettings):
         Json[list[str]],
         Field(
             min_length=1,
-            description="A security group acts as a virtual firewall for your EC2 instances to control incoming and outgoing traffic"
+            description="A security group acts as a virtual firewall for your EC2 instances "
+            "to control incoming and outgoing traffic"
             " (https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-security-groups.html), "
             " this is required to start a new EC2 instance",
         ),
@@ -137,23 +144,27 @@ class EC2InstancesSettings(BaseCustomSettings):
         datetime.timedelta,
         Field(
             description="Time after which an EC2 instance may be drained (10s<=T<=1 minutes, is automatically capped)"
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)",
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types "
+            "for string formatting)",
         ),
     ] = datetime.timedelta(seconds=10)
 
     EC2_INSTANCES_TIME_BEFORE_TERMINATION: Annotated[
         datetime.timedelta,
         Field(
-            description="Time after which an EC2 instance may begin the termination process (0<=T<=59 minutes, is automatically capped)"
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)",
+            description="Time after which an EC2 instance may begin the termination process "
+            "(0<=T<=59 minutes, is automatically capped)"
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types "
+            "for string formatting)",
         ),
     ] = datetime.timedelta(minutes=1)
 
     EC2_INSTANCES_TIME_BEFORE_FINAL_TERMINATION: Annotated[
         datetime.timedelta,
         Field(
             description="Time after which an EC2 instance is terminated after draining"
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)",
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types "
+            "for string formatting)",
         ),
     ] = datetime.timedelta(seconds=30)
 
@@ -167,7 +178,10 @@ class EC2InstancesSettings(BaseCustomSettings):
     EC2_INSTANCES_ATTACHED_IAM_PROFILE: Annotated[
         str,
         Field(
-            description="ARN the EC2 instance should be attached to (example: arn:aws:iam::XXXXX:role/NAME), to disable pass an empty string",
+            description=(
+                "ARN the EC2 instance should be attached to (example: arn:aws:iam::XXXXX:role/NAME), "
+                "to disable pass an empty string"
+            ),
         ),
     ]
 
@@ -210,21 +224,29 @@ class NodesMonitoringSettings(BaseCustomSettings):
     NODES_MONITORING_NODE_LABELS: Annotated[
         list[DockerLabelKey],
         Field(
-            description="autoscaling will only monitor nodes with the given labels (if empty all nodes will be monitored), these labels will be added to the new created nodes by default",
+            description=(
+                "autoscaling will only monitor nodes with the given labels (if empty all nodes will be monitored), "
+                "these labels will be added to the new created nodes by default"
+            ),
         ),
     ]
 
     NODES_MONITORING_SERVICE_LABELS: Annotated[
         list[DockerLabelKey],
         Field(
-            description="autoscaling will only monitor services with the given labels (if empty all services will be monitored)",
+            description=(
+                "autoscaling will only monitor services with the given labels (if empty all services will be monitored)"
+            ),
         ),
     ]
 
     NODES_MONITORING_NEW_NODES_LABELS: Annotated[
         list[DockerLabelKey],
         Field(
-            description="autoscaling will add these labels to any new node it creates (additional to the ones in NODES_MONITORING_NODE_LABELS",
+            description=(
+                "autoscaling will add these labels to any new node it creates "
+                "(additional to the ones in NODES_MONITORING_NODE_LABELS)"
+            ),
         ),
     ]
 
@@ -240,13 +262,19 @@ class DaskMonitoringSettings(BaseCustomSettings):
     DASK_NTHREADS: Annotated[
         NonNegativeInt,
         Field(
-            description="if >0, it overrides the default number of threads per process in the dask-sidecars, (see description in dask-sidecar)",
+            description=(
+                "if >0, it overrides the default number of threads per process in the dask-sidecars, "
+                "(see description in dask-sidecar)"
+            ),
         ),
     ]
     DASK_NTHREADS_MULTIPLIER: Annotated[
         PositiveInt,
         Field(
-            description="if >1, it overrides the default number of threads per process in the dask-sidecars, by multiplying the number of vCPUs with this factor (see description in dask-sidecar)",
+            description=(
+                "if >1, it overrides the default number of threads per process in the dask-sidecars, "
+                "by multiplying the number of vCPUs with this factor (see description in dask-sidecar)"
+            ),
         ),
     ]
 
@@ -282,7 +310,10 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
                 "AUTOSCALING_LOG_FORMAT_LOCAL_DEV_ENABLED",
                 "LOG_FORMAT_LOCAL_DEV_ENABLED",
             ),
-            description="Enables local development log format. WARNING: make sure it is disabled if you want to have structured logs!",
+            description=(
+                "Enables local development log format. WARNING: make sure it is disabled "
+                "if you want to have structured logs!"
+            ),
         ),
     ] = False
 
@@ -291,7 +322,10 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
         Field(
             default_factory=dict,
             validation_alias=AliasChoices("AUTOSCALING_LOG_FILTER_MAPPING", "LOG_FILTER_MAPPING"),
-            description="is a dictionary that maps specific loggers (such as 'uvicorn.access' or 'gunicorn.access') to a list of log message patterns that should be filtered out.",
+            description=(
+                "is a dictionary that maps specific loggers (such as 'uvicorn.access' or 'gunicorn.access') "
+                "to a list of log message patterns that should be filtered out."
+            ),
         ),
     ]
 
@@ -318,8 +352,11 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
     AUTOSCALING_POLL_INTERVAL: Annotated[
         datetime.timedelta,
         Field(
-            description="interval between each resource check "
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)",
+            description=(
+                "interval between each resource check "
+                "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types "
+                "for string formatting)"
+            ),
         ),
     ] = datetime.timedelta(seconds=10)
 
@@ -337,6 +374,11 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
         Field(json_schema_extra={"auto_default_from_env": True}),
     ]
 
+    AUTOSCALING_DOCKER_API_PROXY: Annotated[
+        DockerApiProxysettings | None,
+        Field(json_schema_extra={"auto_default_from_env": True}),
+    ]
+
     AUTOSCALING_PROMETHEUS_INSTRUMENTATION_ENABLED: bool = True
 
     AUTOSCALING_DRAIN_NODES_WITH_LABELS: Annotated[
@@ -382,7 +424,10 @@ def _valid_log_level(cls, value: str) -> str:
     @model_validator(mode="after")
     def _exclude_both_dynamic_computational_mode(self) -> Self:
         if self.AUTOSCALING_DASK is not None and self.AUTOSCALING_NODES_MONITORING is not None:
-            msg = "Autoscaling cannot be set to monitor both computational and dynamic services (both AUTOSCALING_DASK and AUTOSCALING_NODES_MONITORING are currently set!)"
+            msg = (
+                "Autoscaling cannot be set to monitor both computational and dynamic services "
+                "(both AUTOSCALING_DASK and AUTOSCALING_NODES_MONITORING are currently set!)"
+            )
             raise ValueError(msg)
         return self
 

services/autoscaling/src/simcore_service_autoscaling/modules/docker.py

@@ -3,11 +3,14 @@
 
 import aiodocker
 from fastapi import FastAPI
+from servicelib.fastapi.docker import get_remote_docker_client
 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_random_exponential
 
+from ..core.settings import ApplicationSettings
+
 logger = logging.getLogger(__name__)
 
 
@@ -21,8 +24,23 @@ async def ping(self) -> bool:
 
 
 def setup(app: FastAPI) -> None:
+    settings: ApplicationSettings = app.state.settings
+
     async def on_startup() -> None:
-        app.state.docker_client = client = AutoscalingDocker()
+        # Get the remote docker client configured by servicelib
+        if settings.AUTOSCALING_DOCKER_API_PROXY:
+            remote_client = get_remote_docker_client(app)
+
+            # Wrap it with AutoscalingDocker to add the ping method
+            client = AutoscalingDocker(
+                url=remote_client.docker_host,
+                connector=remote_client.connector,
+                session=remote_client.session,
+            )
+        else:
+            # Local docker client
+            client = AutoscalingDocker()
+        app.state.docker_client = client
 
         async for attempt in AsyncRetrying(
             reraise=True,