(⚠️ devops) ♻️✨Adding distributed locking to throttle concurrent saves on nodes (ITISFoundation#3160)

* adds distributed locking per resource per node
* saving states and outputs and pulling states and inputs are now limited per node
* all options are disabled by default

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

Author: GitHK

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

@@ -93,5 +93,8 @@ async def redis_locks_client(
 async def wait_till_redis_responsive(redis_url: Union[URL, str]) -> None:
     client = from_url(f"{redis_url}", encoding="utf-8", decode_responses=True)
 
-    if not await client.ping():
-        raise ConnectionError(f"{redis_url=} not available")
+    try:
+        if not await client.ping():
+            raise ConnectionError(f"{redis_url=} not available")
+    finally:
+        await client.close(close_connection_pool=True)

services/director-v2/requirements/_base.in

@@ -28,4 +28,5 @@ httpx
 networkx
 orjson
 pydantic[dotenv]
+redis
 tenacity

services/director-v2/requirements/_base.txt

@@ -69,6 +69,7 @@ async-timeout==4.0.2
     #   aiohttp
     #   aiopg
     #   aioredis
+    #   redis
 attrs==20.3.0
     # via
     #   -c requirements/../../../packages/service-library/requirements/././constraints.txt
@@ -109,6 +110,8 @@ dask-gateway==2022.6.1
     # via -r requirements/_base.in
 decorator==4.4.2
     # via networkx
+deprecated==1.2.13
+    # via redis
 distributed==2022.6.0
     # via
     #   -r requirements/../../../services/dask-sidecar/requirements/_dask-distributed.txt
@@ -242,6 +245,7 @@ packaging==21.3
     #   -r requirements/../../../services/dask-sidecar/requirements/_dask-distributed.txt
     #   dask
     #   distributed
+    #   redis
 pamqp==2.3.0
     # via aiormq
 partd==1.2.0
@@ -329,6 +333,8 @@ pyyaml==5.4.1
     #   distributed
     #   fastapi
     #   uvicorn
+redis==4.3.4
+    # via -r requirements/_base.in
 requests==2.27.1
     # via
     #   async-asgi-testclient
@@ -444,6 +450,8 @@ watchgod==0.8.2
     # via uvicorn
 websockets==10.1
     # via uvicorn
+wrapt==1.14.1
+    # via deprecated
 yarl==1.7.2
     # via
     #   -r requirements/../../../packages/postgres-database/requirements/_base.in

services/director-v2/requirements/_test.txt

@@ -355,6 +355,7 @@ websocket-client==0.59.0
     #   docker-compose
 wrapt==1.14.1
     # via
+    #   -c requirements/_base.txt
     #   aiobotocore
     #   astroid
 yarl==1.7.2

services/director-v2/src/simcore_service_director_v2/core/application.py

@@ -24,6 +24,7 @@
     director_v0,
     dynamic_services,
     dynamic_sidecar,
+    node_rights,
     rabbitmq,
     remote_debug,
     storage,
@@ -138,6 +139,8 @@ def init_app(settings: Optional[AppSettings] = None) -> FastAPI:
         rabbitmq.setup(app)
         comp_scheduler.setup(app)
 
+    node_rights.setup(app)
+
     if settings.DIRECTOR_V2_TRACING:
         setup_tracing(app, settings.DIRECTOR_V2_TRACING)
 

services/director-v2/src/simcore_service_director_v2/core/errors.py

@@ -19,7 +19,7 @@
 }
 """
 
-from typing import List, Optional
+from typing import Optional
 
 from models_library.errors import ErrorDict
 from models_library.projects import ProjectID
@@ -104,6 +104,10 @@ class ComputationalRunNotFoundError(PydanticErrorMixin, DirectorException):
     msg_template = "Computational run not found"
 
 
+class NodeRightsAcquireError(PydanticErrorMixin, DirectorException):
+    msg_template = "Could not acquire a lock for {docker_node_id} since all {slots} slots are used."
+
+
 #
 # SCHEDULER ERRORS
 #
@@ -133,7 +137,7 @@ def __init__(
         self.project_id = project_id
         self.node_id = node_id
 
-    def get_errors(self) -> List[ErrorDict]:
+    def get_errors(self) -> list[ErrorDict]:
         # default implementation
         return [
             {
@@ -175,15 +179,15 @@ class PortsValidationError(TaskSchedulingError):
     ports in a project's node.
     """
 
-    def __init__(self, project_id: ProjectID, node_id: NodeID, errors: List[ErrorDict]):
+    def __init__(self, project_id: ProjectID, node_id: NodeID, errors: list[ErrorDict]):
         super().__init__(
             project_id,
             node_id,
             msg=f"Node with {len(errors)} ports having invalid values",
         )
         self.errors = errors
 
-    def get_errors(self) -> List[ErrorDict]:
+    def get_errors(self) -> list[ErrorDict]:
         """Returns 'public errors': filters only value_error.port_validation errors for the client.
         The rest only shown as number
         """

services/director-v2/src/simcore_service_director_v2/core/settings.py

@@ -38,6 +38,7 @@
 from settings_library.postgres import PostgresSettings
 from settings_library.r_clone import RCloneSettings
 from settings_library.rabbit import RabbitSettings
+from settings_library.redis import RedisSettings
 from settings_library.tracing import TracingSettings
 from settings_library.utils_logging import MixinLoggingSettings
 from settings_library.utils_service import DEFAULT_FASTAPI_PORT
@@ -300,6 +301,24 @@ class DynamicSidecarSettings(BaseCustomSettings):
         description="exposes the proxy on localhost for debuging and testing",
     )
 
+    DYNAMIC_SIDECAR_DOCKER_NODE_RESOURCE_LIMITS_ENABLED: bool = Field(
+        False,
+        description=(
+            "Limits concurrent service saves for a docker node. Guarantees "
+            "that no more than X services use a resource together."
+        ),
+    )
+    DYNAMIC_SIDECAR_DOCKER_NODE_CONCURRENT_RESOURCE_SLOTS: PositiveInt = Field(
+        2, description="Amount of slots per resource on a node"
+    )
+    DYNAMIC_SIDECAR_DOCKER_NODE_SAVES_LOCK_TIMEOUT_S: PositiveFloat = Field(
+        10,
+        description=(
+            "Lifetime of the lock. Allows the system to recover a lock "
+            "in case of crash, the lock will expire and result as released."
+        ),
+    )
+
     @validator("DYNAMIC_SIDECAR_MOUNT_PATH_DEV", pre=True)
     @classmethod
     def auto_disable_if_production(cls, v, values):
@@ -469,6 +488,8 @@ class AppSettings(BaseCustomSettings, MixinLoggingSettings):
 
     POSTGRES: PGSettings = Field(auto_default_from_env=True)
 
+    REDIS: RedisSettings = Field(auto_default_from_env=True)
+
     DIRECTOR_V2_RABBITMQ: RabbitSettings = Field(auto_default_from_env=True)
 
     TRAEFIK_SIMCORE_ZONE: str = Field("internal_simcore_stack")

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

@@ -352,9 +352,6 @@ class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
         ..., description="service resources used to enforce limits"
     )
 
-    # Below values are used only once and then are nto required, thus optional
-    # after the service is picked up by the scheduler after a reboot these are not required
-    # and can be set to None
     request_dns: Optional[str] = Field(
         None, description="used when configuring the CORS options on the proxy"
     )
@@ -364,6 +361,13 @@ class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
     proxy_service_name: Optional[str] = Field(
         None, description="service name given to the proxy"
     )
+    docker_node_id: Optional[str] = Field(
+        None,
+        description=(
+            "contains node id of the docker node where all services "
+            "and created containers are started"
+        ),
+    )
 
     @classmethod
     def from_http_request(

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

@@ -18,7 +18,6 @@ def get_dynamic_proxy_spec(
     dynamic_sidecar_network_id: str,
     swarm_network_id: str,
     swarm_network_name: str,
-    dynamic_sidecar_node_id: str,
     entrypoint_container_name: str,
     service_port: PositiveInt,
 ) -> Dict[str, Any]:
@@ -98,7 +97,7 @@ def get_dynamic_proxy_spec(
             "Placement": {
                 "Constraints": [
                     "node.platform.os == linux",
-                    f"node.id == {dynamic_sidecar_node_id}",
+                    f"node.id == {scheduler_data.docker_node_id}",
                 ]
             },
             "Resources": {