♻️🐛 director-v2 sidecar tracking fixes (⚠️ devops) (ITISFoundation#3272)

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

Author: GitHK

services/director-v2/src/simcore_service_director_v2/cli.py

@@ -18,12 +18,12 @@
 from .core.application import create_base_app
 from .core.settings import AppSettings
 from .meta import PROJECT_NAME
-from .models.schemas.dynamic_services import DynamicSidecarNames
+from .models.schemas.dynamic_services import DynamicSidecarNamesHelper
 from .modules import db, director_v0, dynamic_sidecar
 from .modules.db.repositories.projects import ProjectsRepository
 from .modules.director_v0 import DirectorV0Client
 from .modules.dynamic_sidecar import api_client
-from .modules.dynamic_sidecar.scheduler._utils import fetch_repo_outside_of_request
+from .modules.dynamic_sidecar.scheduler._utils import get_repository
 from .modules.projects_networks import requires_dynamic_sidecar
 
 DEFAULT_NODE_SAVE_RETRY: Final[int] = 3
@@ -52,7 +52,7 @@ async def _initialized_app() -> AsyncIterator[FastAPI]:
 def _get_dynamic_sidecar_endpoint(
     settings: AppSettings, node_id: NodeIDStr
 ) -> AnyHttpUrl:
-    dynamic_sidecar_names = DynamicSidecarNames.make(UUID(node_id))
+    dynamic_sidecar_names = DynamicSidecarNamesHelper.make(UUID(node_id))
     hostname = dynamic_sidecar_names.service_name_dynamic_sidecar
     port = settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR.DYNAMIC_SIDECAR_PORT
     return parse_obj_as(AnyHttpUrl, f"http://{hostname}:{port}")  # NOSONAR
@@ -80,7 +80,7 @@ async def _save_node_state(
 
 async def _async_project_save_state(project_id: ProjectID, retry_save: int) -> None:
     async with _initialized_app() as app:
-        projects_repository: ProjectsRepository = fetch_repo_outside_of_request(
+        projects_repository: ProjectsRepository = get_repository(
             app, ProjectsRepository
         )
         project_at_db = await projects_repository.get_project(project_id)

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

@@ -305,7 +305,12 @@ class DynamicSidecarSettings(BaseCustomSettings):
         False,
         description=(
             "Limits concurrent service saves for a docker node. Guarantees "
-            "that no more than X services use a resource together."
+            "that no more than X services use a resource together. "
+            "NOTE: A node can end up with all the services from a single study. "
+            "When the study is closed/opened all the services will try to "
+            "upload/download their data. This causes a lot of disk "
+            "and network stress (especially for low power nodes like in AWS). "
+            "Some nodes collapse under load or behave unexpectedly."
         ),
     )
     DYNAMIC_SIDECAR_DOCKER_NODE_CONCURRENT_RESOURCE_SLOTS: PositiveInt = Field(
@@ -449,6 +454,14 @@ class AppSettings(BaseCustomSettings, MixinLoggingSettings):
     )
     DIRECTOR_V2_DEV_FEATURES_ENABLED: bool = False
 
+    DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED: bool = Field(
+        False,
+        description=(
+            "Under development feature. If enabled state "
+            "is saved using rclone docker volumes."
+        ),
+    )
+
     # for passing self-signed certificate to spawned services
     # TODO: fix these variables once the timeout-minutes: 30 is able to start dynamic services
     DIRECTOR_V2_SELF_SIGNED_SSL_SECRET_ID: str = ""

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

@@ -4,6 +4,7 @@
 from typing import Any, Mapping, Optional
 from uuid import UUID, uuid4
 
+from models_library.basic_types import PortInt
 from models_library.projects_nodes_io import NodeID
 from models_library.service_settings_labels import (
     DynamicSidecarServiceLabels,
@@ -12,15 +13,7 @@
 )
 from models_library.services import RunID
 from models_library.services_resources import ServiceResourcesDict
-from pydantic import (
-    AnyHttpUrl,
-    BaseModel,
-    Extra,
-    Field,
-    PositiveInt,
-    constr,
-    parse_obj_as,
-)
+from pydantic import AnyHttpUrl, BaseModel, Extra, Field, constr, parse_obj_as
 from servicelib.error_codes import ErrorCodeStr
 
 from ..constants import (
@@ -39,6 +32,7 @@
 DockerId = constr(max_length=25, regex=r"[A-Za-z0-9]{25}")
 ServiceId = DockerId
 NetworkId = DockerId
+ServiceName = constr(strip_whitespace=True, min_length=2)
 
 logger = logging.getLogger()
 
@@ -143,25 +137,11 @@ def mark_removed(self) -> None:
 
 
 class DynamicSidecar(BaseModel):
-    run_id: RunID = Field(
-        default_factory=uuid4,
-        description=(
-            "Used to discriminate between dynamic-sidecar docker resources "
-            "generated during different runs. Sometimes artifacts remain in the"
-            "system after an error. This helps avoiding collisions."
-            "For now used by anonymous volumes involved in data sharing"
-        ),
-    )
-
     status: Status = Field(
         Status.create_as_initially_ok(),
         description="status of the service sidecar also with additional information",
     )
 
-    hostname: str = Field(..., description="docker hostname for this service")
-
-    port: PositiveInt = Field(8000, description="dynamic-sidecar port")
-
     is_available: bool = Field(
         False,
         scription=(
@@ -224,6 +204,18 @@ def compose_spec_submitted(self) -> bool:
         ),
     )
 
+    wait_for_manual_intervention_after_error: bool = Field(
+        False,
+        description=(
+            "Marks the sidecar as untouchable since there was an error and "
+            "important data might be lost. awaits for manual intervention."
+        ),
+    )
+    were_state_and_outputs_saved: bool = Field(
+        False,
+        description="set True if the dy-sidecar saves the state and uploads the outputs",
+    )
+
     # below had already been validated and
     # used only to start the proxy
     dynamic_sidecar_id: Optional[ServiceId] = Field(
@@ -239,34 +231,19 @@ def compose_spec_submitted(self) -> bool:
         None, description="used for starting the proxy"
     )
 
-    @property
-    def can_save_state(self) -> bool:
-        """
-        Keeps track of the current state of the application, if it was starte successfully
-        the state of the service can be saved when stopping the service
-        """
-        # TODO: implement when adding save status hooks
-        return False
-
-    # consider adding containers for healthchecks but this is more difficult and it depends on each service
-
-    @property
-    def endpoint(self) -> AnyHttpUrl:
-        """endpoint where all the services are exposed"""
-        return parse_obj_as(
-            AnyHttpUrl, f"http://{self.hostname}:{self.port}"  # NOSONAR
-        )
+    docker_node_id: Optional[str] = Field(
+        None,
+        description=(
+            "contains node id of the docker node where all services "
+            "and created containers are started"
+        ),
+    )
 
-    @property
-    def are_containers_ready(self) -> bool:
-        """returns: True if all containers are in running state"""
-        return all(
-            docker_container_inspect.status == DockerStatus.RUNNING
-            for docker_container_inspect in self.containers_inspect
-        )
+    class Config:
+        validate_assignment = True
 
 
-class DynamicSidecarNames(BaseModel):
+class DynamicSidecarNamesHelper(BaseModel):
     """
     Service naming schema:
     NOTE: name is max 63 characters
@@ -304,7 +281,7 @@ class DynamicSidecarNames(BaseModel):
     )
 
     @classmethod
-    def make(cls, node_uuid: UUID) -> "DynamicSidecarNames":
+    def make(cls, node_uuid: UUID) -> "DynamicSidecarNamesHelper":
         return cls(
             service_name_dynamic_sidecar=assemble_service_name(
                 DYNAMIC_SIDECAR_SERVICE_PREFIX, node_uuid
@@ -318,10 +295,28 @@ def make(cls, node_uuid: UUID) -> "DynamicSidecarNames":
 
 
 class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
-    service_name: constr(strip_whitespace=True, min_length=2) = Field(
+    service_name: ServiceName = Field(
         ...,
         description="Name of the current dynamic-sidecar being observed",
     )
+    run_id: RunID = Field(
+        default_factory=uuid4,
+        description=(
+            "Uniquely identify the dynamic sidecar session (a.k.a. 2 "
+            "subsequent exact same services will have a different run_id)"
+        ),
+    )
+    hostname: str = Field(
+        ..., description="dy-sidecar's service hostname (provided by docker-swarm)"
+    )
+    port: PortInt = Field(8000, description="dynamic-sidecar port")
+
+    @property
+    def endpoint(self) -> AnyHttpUrl:
+        """endpoint where all the services are exposed"""
+        return parse_obj_as(
+            AnyHttpUrl, f"http://{self.hostname}:{self.port}"  # NOSONAR
+        )
 
     dynamic_sidecar: DynamicSidecar = Field(
         ...,
@@ -340,7 +335,7 @@ class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
         description="required for Traefik to correctly route requests to the spawned container",
     )
 
-    service_port: PositiveInt = Field(
+    service_port: PortInt = Field(
         TEMPORARY_PORT_NUMBER,
         description=(
             "port where the service is exposed defined by the service; "
@@ -353,37 +348,31 @@ class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
         ..., description="service resources used to enforce limits"
     )
 
-    request_dns: Optional[str] = Field(
-        None, description="used when configuring the CORS options on the proxy"
-    )
-    request_scheme: Optional[str] = Field(
-        None, description="used when configuring the CORS options on the proxy"
-    )
-    proxy_service_name: Optional[str] = Field(
-        None, description="service name given to the proxy"
+    request_dns: str = Field(
+        ..., description="used when configuring the CORS options on 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"
-        ),
+    request_scheme: str = Field(
+        ..., description="used when configuring the CORS options on the proxy"
     )
+    proxy_service_name: str = Field(None, description="service name given to the proxy")
 
     @classmethod
     def from_http_request(
         # pylint: disable=too-many-arguments
         cls,
         service: "DynamicServiceCreate",
         simcore_service_labels: SimcoreServiceLabels,
-        port: Optional[int],
-        request_dns: Optional[str] = None,
-        request_scheme: Optional[str] = None,
+        port: PortInt,
+        request_dns: str,
+        request_scheme: str,
+        run_id: Optional[UUID] = None,
     ) -> "SchedulerData":
-        dynamic_sidecar_names = DynamicSidecarNames.make(service.node_uuid)
+        names_helper = DynamicSidecarNamesHelper.make(service.node_uuid)
 
         obj_dict = dict(
-            service_name=dynamic_sidecar_names.service_name_dynamic_sidecar,
+            service_name=names_helper.service_name_dynamic_sidecar,
+            hostname=names_helper.service_name_dynamic_sidecar,
+            port=port,
             node_uuid=service.node_uuid,
             project_id=service.project_id,
             user_id=service.user_id,
@@ -394,16 +383,15 @@ def from_http_request(
             compose_spec=json.dumps(simcore_service_labels.compose_spec),
             container_http_entry=simcore_service_labels.container_http_entry,
             restart_policy=simcore_service_labels.restart_policy,
-            dynamic_sidecar_network_name=dynamic_sidecar_names.dynamic_sidecar_network_name,
-            simcore_traefik_zone=dynamic_sidecar_names.simcore_traefik_zone,
+            dynamic_sidecar_network_name=names_helper.dynamic_sidecar_network_name,
+            simcore_traefik_zone=names_helper.simcore_traefik_zone,
             request_dns=request_dns,
             request_scheme=request_scheme,
-            proxy_service_name=dynamic_sidecar_names.proxy_service_name,
-            dynamic_sidecar=dict(
-                hostname=dynamic_sidecar_names.service_name_dynamic_sidecar,
-                port=port,
-            ),
+            proxy_service_name=names_helper.proxy_service_name,
+            dynamic_sidecar={},
         )
+        if run_id:
+            obj_dict["run_id"] = run_id
         return cls.parse_obj(obj_dict)
 
     @classmethod

services/director-v2/src/simcore_service_director_v2/modules/catalog.py

@@ -14,8 +14,6 @@
 
 logger = logging.getLogger(__name__)
 
-_MINUTE = 60
-
 
 def setup(app: FastAPI, settings: CatalogSettings) -> None:
     if not settings:

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_public.py

@@ -402,7 +402,7 @@ async def get_dynamic_sidecar_service_health(
     app: FastAPI, scheduler_data: SchedulerData
 ) -> None:
     api_client = get_dynamic_sidecar_client(app)
-    service_endpoint = scheduler_data.dynamic_sidecar.endpoint
+    service_endpoint = scheduler_data.endpoint
 
     # update service health
     is_healthy = await api_client.is_healthy(service_endpoint)

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

@@ -254,19 +254,27 @@ def _make_pending() -> tuple[ServiceState, str]:
     return service_state, message
 
 
-async def is_dynamic_sidecar_stack_missing(
+async def _get_dynamic_sidecar_stack_services(
     node_uuid: NodeID, dynamic_sidecar_settings: DynamicSidecarSettings
-) -> bool:
-    """Check if the proxy and the dynamic-sidecar are absent"""
+) -> list[Mapping]:
     filters = {
         "label": [
             f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}",
             f"uuid={node_uuid}",
         ]
     }
     async with docker_client() as client:
-        stack_services = await client.services.list(filters=filters)
-        return len(stack_services) == 0
+        return await client.services.list(filters=filters)
+
+
+async def is_dynamic_sidecar_stack_missing(
+    node_uuid: NodeID, dynamic_sidecar_settings: DynamicSidecarSettings
+) -> bool:
+    """Check if the proxy and the dynamic-sidecar are absent"""
+    stack_services = await _get_dynamic_sidecar_stack_services(
+        node_uuid, dynamic_sidecar_settings
+    )
+    return len(stack_services) == 0
 
 
 async def are_all_services_present(
@@ -275,20 +283,13 @@ async def are_all_services_present(
     """
     The dynamic-sidecar stack always expects to have 2 running services
     """
-    async with docker_client() as client:
-        stack_services = await client.services.list(
-            filters={
-                "label": [
-                    f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}",
-                    f"uuid={node_uuid}",
-                ]
-            }
-        )
-        if len(stack_services) != 2:
-            log.warning("Expected 2 services found %s", stack_services)
-            return False
+    stack_services = await _get_dynamic_sidecar_stack_services(
+        node_uuid, dynamic_sidecar_settings
+    )
+    if len(stack_services) != 2:
+        return False
 
-        return True
+    return True
 
 
 async def remove_dynamic_sidecar_stack(
@@ -437,8 +438,12 @@ def _count_containers(item: dict[str, Any]) -> int:
 async def try_to_remove_network(network_name: str) -> None:
     async with docker_client() as client:
         network = await client.networks.get(network_name)
+
+        # if a project network for the current project has no more
+        # containers attached to it (because the last service which
+        # was using it was removed), also removed the network
         try:
-            return await network.delete()
+            await network.delete()
         except aiodocker.exceptions.DockerError:
             log.warning("Could not remove network %s", network_name)