✨ add EFS mount option to sidecar

.env-devel

@@ -66,6 +66,7 @@ DIRECTOR_REGISTRY_CACHING=True
 
 EFS_DNS_NAME=fs-xxx.efs.us-east-1.amazonaws.com
 EFS_MOUNTED_PATH=/tmp/efs
+EFS_ENABLED_FOR_USERS={}
 
 # DIRECTOR_V2 ----
 COMPUTATIONAL_BACKEND_DEFAULT_CLUSTER_AUTH='{"type":"tls","tls_ca_file":"/home/scu/.dask/dask-crt.pem","tls_client_cert":"/home/scu/.dask/dask-crt.pem","tls_client_key":"/home/scu/.dask/dask-key.pem"}'

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/efs_guardian/efs_manager.py

@@ -23,6 +23,7 @@ async def create_project_specific_data_dir(
     *,
     project_id: ProjectID,
     node_id: NodeID,
+    storage_directory_name: str,
 ) -> Path:
     output: Path = await rabbitmq_rpc_client.request(
         EFS_GUARDIAN_RPC_NAMESPACE,

packages/settings-library/src/settings_library/efs.py

@@ -0,0 +1,48 @@
+from pathlib import Path
+
+from pydantic import Field
+
+from .base import BaseCustomSettings
+
+
+class AwsEfsSettings(BaseCustomSettings):
+    EFS_DNS_NAME: str = Field(
+        description="AWS Elastic File System DNS name",
+        example="fs-xxx.efs.us-east-1.amazonaws.com",
+    )
+    EFS_PROJECT_SPECIFIC_DATA_DIRECTORY: str = Field(default="project-specific-data")
+    EFS_MOUNTED_PATH: Path = Field(
+        default=Path("/data/efs"),
+        description="This is the path where EFS is mounted to the EC2 machine",
+    )
+    EFS_ENABLED_FOR_USERS: set[int] = Field(
+        description="This is temporary solution so we can enable it for specific users for testing purpose",
+        example={1},
+    )
+
+
+NFS_PROTOCOL = "4.1"
+READ_SIZE = "1048576"
+WRITE_SIZE = "1048576"
+RECOVERY_MODE = "hard"
+NFS_REQUEST_TIMEOUT = "600"
+NUMBER_OF_RETRANSMISSIONS = "2"
+PORT_MODE = "noresvport"
+
+"""
+`sudo mount -t nfs -o nfsvers=4.1,rsize=1048576,wsize=1048576,hard,timeo=600,retrans=2,noresvport`
+
+Explanation:
+
+nfsvers=4.1: Specifies the NFS protocol version to use; here, it is version 4.1, which supports improved security features and performance optimizations over earlier versions.
+
+rsize=1048576 and wsize=1048576: Set the read and write buffer sizes in bytes, respectively. Here, both are set to 1,048,576 bytes (1 MB). Larger buffer sizes can improve performance over high-latency networks by allowing more data to be transferred with each read or write request.
+
+hard: Specifies the recovery behavior of the NFS client. If the NFS server becomes unreachable, the NFS client will retry the request until the server becomes available again. The alternative is soft, where the NFS client gives up after a certain number of retries, potentially leading to data corruption or loss.
+
+timeo=600: Sets the timeout value for NFS requests in deciseconds (tenths of a second). Here, 600 deciseconds means 60 seconds. This is how long the NFS client will wait for a response from the server before retrying or failing.
+
+retrans=2: Sets the number of retransmissions for each NFS request if a response is not received before the timeout. Here, it will retry each request twice.
+
+noresvport: Normally, NFS uses a reserved port (number below 1024) for communicating, which requires root privileges on the client side. noresvport allows using non-reserved ports, which can be helpful in environments where clients don't have root privileges.
+"""

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

@@ -11,6 +11,7 @@
 )
 from pydantic import Field, PositiveInt, validator
 from settings_library.base import BaseCustomSettings
+from settings_library.efs import AwsEfsSettings
 from settings_library.r_clone import RCloneSettings as SettingsLibraryRCloneSettings
 from settings_library.utils_logging import MixinLoggingSettings
 from settings_library.utils_service import DEFAULT_FASTAPI_PORT
@@ -124,6 +125,10 @@ class DynamicSidecarSettings(BaseCustomSettings, MixinLoggingSettings):
 
     DYNAMIC_SIDECAR_R_CLONE_SETTINGS: RCloneSettings = Field(auto_default_from_env=True)
 
+    DYNAMIC_SIDECAR_EFS_SETTINGS: AwsEfsSettings | None = Field(
+        auto_default_from_env=True
+    )
+
     DYNAMIC_SIDECAR_PLACEMENT_SETTINGS: PlacementSettings = Field(
         auto_default_from_env=True
     )

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

@@ -16,6 +16,8 @@
 from models_library.service_settings_labels import SimcoreServiceSettingsLabel
 from models_library.utils.json_serialization import json_dumps
 from pydantic import ByteSize, parse_obj_as
+from servicelib.rabbitmq import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.efs_guardian import efs_manager
 from servicelib.utils import unused_port
 from settings_library.node_ports import StorageAuthSettings
 
@@ -196,13 +198,14 @@ def get_prometheus_monitoring_networks(
     )
 
 
-def _get_mounts(
+async def _get_mounts(
     *,
     scheduler_data: SchedulerData,
     dynamic_sidecar_settings: DynamicSidecarSettings,
     dynamic_services_scheduler_settings: DynamicServicesSchedulerSettings,
     app_settings: AppSettings,
     has_quota_support: bool,
+    rpc_client: RabbitMQRPCClient,
 ) -> list[dict[str, Any]]:
     mounts: list[dict[str, Any]] = [
         # docker socket needed to use the docker api
@@ -252,10 +255,41 @@ def _get_mounts(
                 volume_size_limit=volume_size_limits.get(f"{path_to_mount}"),
             )
         )
+
+    # We check whether user has access to EFS feature
+    use_efs = False
+    efs_settings = dynamic_sidecar_settings.DYNAMIC_SIDECAR_EFS_SETTINGS
+    if efs_settings and scheduler_data.user_id in efs_settings.EFS_ENABLED_FOR_USERS:
+        use_efs = True
+
     # state paths now get mounted via different driver and are synced to s3 automatically
     for path_to_mount in scheduler_data.paths_mapping.state_paths:
+        if use_efs:
+            assert dynamic_sidecar_settings.DYNAMIC_SIDECAR_EFS_SETTINGS  # nosec
+
+            _storage_directory_name = DynamicSidecarVolumesPathsResolver.volume_name(
+                path_to_mount
+            ).strip("_")
+            await efs_manager.create_project_specific_data_dir(
+                rpc_client,
+                project_id=scheduler_data.project_id,
+                node_id=scheduler_data.node_uuid,
+                storage_directory_name=_storage_directory_name,
+            )
+            mounts.append(
+                DynamicSidecarVolumesPathsResolver.mount_efs(
+                    swarm_stack_name=dynamic_services_scheduler_settings.SWARM_STACK_NAME,
+                    path=path_to_mount,
+                    node_uuid=scheduler_data.node_uuid,
+                    run_id=scheduler_data.run_id,
+                    project_id=scheduler_data.project_id,
+                    user_id=scheduler_data.user_id,
+                    efs_settings=dynamic_sidecar_settings.DYNAMIC_SIDECAR_EFS_SETTINGS,
+                    storage_directory_name=_storage_directory_name,
+                )
+            )
         # for now only enable this with dev features enabled
-        if app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
+        elif app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
             mounts.append(
                 DynamicSidecarVolumesPathsResolver.mount_r_clone(
                     swarm_stack_name=dynamic_services_scheduler_settings.SWARM_STACK_NAME,
@@ -351,7 +385,7 @@ def _get_ports(
     return ports
 
 
-def get_dynamic_sidecar_spec(  # pylint:disable=too-many-arguments# noqa: PLR0913
+async def get_dynamic_sidecar_spec(  # pylint:disable=too-many-arguments# noqa: PLR0913
     scheduler_data: SchedulerData,
     dynamic_sidecar_settings: DynamicSidecarSettings,
     dynamic_services_scheduler_settings: DynamicServicesSchedulerSettings,
@@ -364,6 +398,7 @@ def get_dynamic_sidecar_spec(  # pylint:disable=too-many-arguments# noqa: PLR091
     hardware_info: HardwareInfo | None,
     metrics_collection_allowed: bool,
     telemetry_enabled: bool,
+    rpc_client: RabbitMQRPCClient,
 ) -> AioDockerServiceSpec:
     """
     The dynamic-sidecar is responsible for managing the lifecycle
@@ -375,12 +410,13 @@ def get_dynamic_sidecar_spec(  # pylint:disable=too-many-arguments# noqa: PLR091
     """
     compose_namespace = get_compose_namespace(scheduler_data.node_uuid)
 
-    mounts = _get_mounts(
+    mounts = await _get_mounts(
         scheduler_data=scheduler_data,
         dynamic_services_scheduler_settings=dynamic_services_scheduler_settings,
         dynamic_sidecar_settings=dynamic_sidecar_settings,
         app_settings=app_settings,
         has_quota_support=has_quota_support,
+        rpc_client=rpc_client,
     )
 
     ports = _get_ports(

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

@@ -18,7 +18,7 @@
 from models_library.service_settings_labels import SimcoreServiceSettingsLabel
 from models_library.services import RunID
 from models_library.utils.json_serialization import json_dumps
-from servicelib.rabbitmq import RabbitMQClient
+from servicelib.rabbitmq import RabbitMQClient, RabbitMQRPCClient
 from simcore_postgres_database.models.comp_tasks import NodeClass
 
 from .....core.dynamic_services_settings import DynamicServicesSettings
@@ -222,9 +222,11 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
             user_id=scheduler_data.user_id, product_name=scheduler_data.product_name
         )
 
+        rpc_client: RabbitMQRPCClient = app.state.rabbitmq_rpc_client
+
         # WARNING: do NOT log, this structure has secrets in the open
         # If you want to log, please use an obfuscator
-        dynamic_sidecar_service_spec_base: AioDockerServiceSpec = get_dynamic_sidecar_spec(
+        dynamic_sidecar_service_spec_base: AioDockerServiceSpec = await get_dynamic_sidecar_spec(
             scheduler_data=scheduler_data,
             dynamic_sidecar_settings=dynamic_sidecar_settings,
             dynamic_services_scheduler_settings=dynamic_services_scheduler_settings,
@@ -236,6 +238,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
             allow_internet_access=allow_internet_access,
             metrics_collection_allowed=metrics_collection_allowed,
             telemetry_enabled=is_telemetry_enabled,
+            rpc_client=rpc_client,
         )
 
         catalog_client = CatalogClient.instance(app)

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

@@ -7,6 +7,16 @@
 from models_library.services import RunID
 from models_library.users import UserID
 from servicelib.docker_constants import PREFIX_DYNAMIC_SIDECAR_VOLUMES
+from settings_library.efs import (
+    NFS_PROTOCOL,
+    NFS_REQUEST_TIMEOUT,
+    NUMBER_OF_RETRANSMISSIONS,
+    PORT_MODE,
+    READ_SIZE,
+    RECOVERY_MODE,
+    WRITE_SIZE,
+    AwsEfsSettings,
+)
 from settings_library.r_clone import S3Provider
 
 from ...core.dynamic_services_settings.sidecar import RCloneSettings
@@ -75,6 +85,23 @@ def _get_s3_volume_driver_config(
     return driver_config
 
 
+def _get_efs_volume_driver_config(
+    efs_settings: AwsEfsSettings,
+    project_id: ProjectID,
+    node_uuid: NodeID,
+    storage_directory_name: str,
+) -> dict[str, Any]:
+    assert "/" not in storage_directory_name  # nosec
+    driver_config: dict[str, Any] = {
+        "Options": {
+            "type": "nfs",
+            "o": f"addr={efs_settings.EFS_DNS_NAME},rw,nfsvers={NFS_PROTOCOL},rsize={READ_SIZE},wsize={WRITE_SIZE},{RECOVERY_MODE},timeo={NFS_REQUEST_TIMEOUT},retrans={NUMBER_OF_RETRANSMISSIONS},{PORT_MODE}",
+            "device": f":/{efs_settings.EFS_PROJECT_SPECIFIC_DATA_DIRECTORY}/{project_id}/{node_uuid}/{storage_directory_name}",
+        },
+    }
+    return driver_config
+
+
 class DynamicSidecarVolumesPathsResolver:
     BASE_PATH: Path = Path("/dy-volumes")
 
@@ -85,7 +112,7 @@ def target(cls, path: Path) -> str:
         return f"{target_path}"
 
     @classmethod
-    def _volume_name(cls, path: Path) -> str:
+    def volume_name(cls, path: Path) -> str:
         return f"{path}".replace(os.sep, "_")
 
     @classmethod
@@ -104,7 +131,7 @@ def source(cls, path: Path, node_uuid: NodeID, run_id: RunID) -> str:
         # NOTE: issues can occur when the paths of the mounted outputs, inputs
         # and state folders are very long and share the same subdirectory path.
         # Reversing volume name to prevent these issues from happening.
-        reversed_volume_name = cls._volume_name(path)[::-1]
+        reversed_volume_name = cls.volume_name(path)[::-1]
         unique_name = f"{PREFIX_DYNAMIC_SIDECAR_VOLUMES}_{run_id}_{node_uuid}_{reversed_volume_name}"
         return unique_name[:255]
 
@@ -217,7 +244,41 @@ def mount_r_clone(
                     r_clone_settings=r_clone_settings,
                     project_id=project_id,
                     node_uuid=node_uuid,
-                    storage_directory_name=cls._volume_name(path).strip("_"),
+                    storage_directory_name=cls.volume_name(path).strip("_"),
+                ),
+            },
+        }
+
+    @classmethod
+    def mount_efs(
+        cls,
+        swarm_stack_name: str,
+        path: Path,
+        node_uuid: NodeID,
+        run_id: RunID,
+        project_id: ProjectID,
+        user_id: UserID,
+        efs_settings: AwsEfsSettings,
+        storage_directory_name: str,
+    ) -> dict[str, Any]:
+        return {
+            "Source": cls.source(path, node_uuid, run_id),
+            "Target": cls.target(path),
+            "Type": "volume",
+            "VolumeOptions": {
+                "Labels": {
+                    "source": cls.source(path, node_uuid, run_id),
+                    "run_id": f"{run_id}",
+                    "node_uuid": f"{node_uuid}",
+                    "study_id": f"{project_id}",
+                    "user_id": f"{user_id}",
+                    "swarm_stack_name": swarm_stack_name,
+                },
+                "DriverConfig": _get_efs_volume_driver_config(
+                    efs_settings=efs_settings,
+                    project_id=project_id,
+                    node_uuid=node_uuid,
+                    storage_directory_name=storage_directory_name,
                 ),
             },
         }

services/director-v2/tests/unit/with_dbs/test_modules_dynamic_sidecar_docker_service_specs.py

@@ -6,6 +6,7 @@
 import json
 from collections.abc import Mapping
 from typing import Any, cast
+from unittest.mock import Mock
 
 import pytest
 import respx
@@ -408,7 +409,7 @@ def expected_dynamic_sidecar_spec(
     }
 
 
-def test_get_dynamic_proxy_spec(
+async def test_get_dynamic_proxy_spec(
     mocked_catalog_service_api: respx.MockRouter,
     minimal_app: FastAPI,
     scheduler_data: SchedulerData,
@@ -436,7 +437,7 @@ def test_get_dynamic_proxy_spec(
     for count in range(1, 11):  # loop to check it does not repeat copies
         print(f"{count:*^50}")
 
-        dynamic_sidecar_spec: AioDockerServiceSpec = get_dynamic_sidecar_spec(
+        dynamic_sidecar_spec: AioDockerServiceSpec = await get_dynamic_sidecar_spec(
             scheduler_data=scheduler_data,
             dynamic_sidecar_settings=dynamic_sidecar_settings,
             dynamic_services_scheduler_settings=dynamic_services_scheduler_settings,
@@ -448,6 +449,7 @@ def test_get_dynamic_proxy_spec(
             allow_internet_access=False,
             metrics_collection_allowed=True,
             telemetry_enabled=True,
+            rpc_client=Mock(),
         )
 
         exclude_keys: Mapping[int | str, Any] = {
@@ -530,7 +532,7 @@ async def test_merge_dynamic_sidecar_specs_with_user_specific_specs(
     hardware_info: HardwareInfo,
     fake_service_specifications: dict[str, Any],
 ):
-    dynamic_sidecar_spec: AioDockerServiceSpec = get_dynamic_sidecar_spec(
+    dynamic_sidecar_spec: AioDockerServiceSpec = await get_dynamic_sidecar_spec(
         scheduler_data=scheduler_data,
         dynamic_sidecar_settings=dynamic_sidecar_settings,
         dynamic_services_scheduler_settings=dynamic_services_scheduler_settings,
@@ -542,6 +544,7 @@ async def test_merge_dynamic_sidecar_specs_with_user_specific_specs(
         allow_internet_access=False,
         metrics_collection_allowed=True,
         telemetry_enabled=True,
+        rpc_client=Mock(),
     )
     assert dynamic_sidecar_spec
     dynamic_sidecar_spec_dict = dynamic_sidecar_spec.dict()

services/docker-compose.yml

@@ -300,6 +300,10 @@ services:
       R_CLONE_OPTION_TRANSFERS: ${R_CLONE_OPTION_TRANSFERS}
       R_CLONE_PROVIDER: ${R_CLONE_PROVIDER}
 
+      EFS_DNS_NAME: ${EFS_DNS_NAME}
+      EFS_MOUNTED_PATH: ${EFS_MOUNTED_PATH}
+      EFS_ENABLED_FOR_USERS: ${EFS_ENABLED_FOR_USERS}
+
       RABBIT_HOST: ${RABBIT_HOST}
       RABBIT_PASSWORD: ${RABBIT_PASSWORD}
       RABBIT_PORT: ${RABBIT_PORT}
@@ -358,6 +362,7 @@ services:
       RABBIT_USER: ${RABBIT_USER}
       EFS_DNS_NAME: ${EFS_DNS_NAME}
       EFS_MOUNTED_PATH: ${EFS_MOUNTED_PATH}
+      EFS_ENABLED_FOR_USERS: ${EFS_ENABLED_FOR_USERS}
 
   invitations:
     image: ${DOCKER_REGISTRY:-itisfoundation}/invitations:${DOCKER_IMAGE_TAG:-latest}