🐛 persist dynamic-sidecar state to docker service labels (ITISFoundation#2961)

* serializing path

* initial commit

* task updates

* fixed tests

* fixed boot

* pytest

* remove reraise

* added new settings

* no failiure if service was not found

* bumped timeout

* added test to ensure compare works

* update tests and properties

* updated docstring

* update comment

* fix notes

* moved test to correct place

* simplified call

* refactor test

* more test refactoring

* using parse_raw

* add extra enfrocement

* fix pylint

* removed json encoders

* fixed comments and styles

* renamed label

* removed notes

* fix comment

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

Author: GitHK

packages/models-library/tests/test_service_settings_labels.py

@@ -2,6 +2,7 @@
 # pylint:disable=unused-argument
 # pylint:disable=redefined-outer-name
 
+import json
 from collections import namedtuple
 from copy import deepcopy
 from pprint import pformat
@@ -123,6 +124,13 @@ def test_path_mappings_none_state_paths() -> None:
         PathMappingsLabel(**sample_data)
 
 
+def test_path_mappings_json_encoding() -> None:
+    example = PathMappingsLabel.Config.schema_extra["example"]
+    path_mappings = PathMappingsLabel.parse_obj(example)
+    print(path_mappings)
+    assert json.loads(path_mappings.json()) == example
+
+
 def test_simcore_services_labels_compose_spec_null_container_http_entry_provided() -> None:
     sample_data = deepcopy(SimcoreServiceLabels.Config.schema_extra["examples"][2])
     assert sample_data["simcore.service.container-http-entrypoint"]

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

@@ -221,6 +221,12 @@ class DynamicSidecarSettings(BaseCustomSettings):
             "timeout for attaching/detaching project networks to/from a container"
         ),
     )
+    DYNAMIC_SIDECAR_VOLUMES_REMOVAL_TIMEOUT_S: PositiveFloat = Field(
+        1.0 * MINS,
+        description=(
+            "time to wait before giving up on removing dynamic-sidecar's volumes"
+        ),
+    )
 
     TRAEFIK_SIMCORE_ZONE: str = Field(
         ...,

services/director-v2/src/simcore_service_director_v2/models/schemas/constants.py

@@ -3,6 +3,10 @@
 DYNAMIC_SIDECAR_SERVICE_PREFIX = "dy-sidecar"
 DYNAMIC_PROXY_SERVICE_PREFIX = "dy-proxy"
 
+# label storing scheduler_data to allow service
+# monitoring recovery after director-v2 reboots
+DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL = "io.simcore.scheduler-data"
+
 # This matches registries by:
 # - local
 # - itisfoundation

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

@@ -1,23 +1,21 @@
-import datetime
 import json
 import logging
 from asyncio import Lock
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional
 from uuid import UUID
 
-from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.service_settings_labels import (
     DynamicSidecarServiceLabels,
     PathMappingsLabel,
-    RestartPolicy,
     SimcoreServiceLabels,
 )
 from pydantic import BaseModel, Extra, Field, PositiveInt, PrivateAttr, constr
 
 from ..constants import (
     DYNAMIC_PROXY_SERVICE_PREFIX,
+    DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL,
     DYNAMIC_SIDECAR_SERVICE_PREFIX,
     REGEX_DY_SERVICE_PROXY,
     REGEX_DY_SERVICE_SIDECAR,
@@ -28,9 +26,9 @@
 
 MAX_ALLOWED_SERVICE_NAME_LENGTH: int = 63
 
-SHA256 = constr(max_length=64, regex=r"\b[A-Fa-f0-9]{64}\b")
-ServiceId = SHA256
-NetworkId = SHA256
+DockerId = constr(max_length=25, regex=r"[A-Za-z0-9]{25}")
+ServiceId = DockerId
+NetworkId = DockerId
 
 logger = logging.getLogger()
 
@@ -94,11 +92,6 @@ class DockerContainerInspect(BaseModel):
     name: str = Field(..., description="docker name of the container")
     id: str = Field(..., description="docker id of the container")
 
-    last_updated: datetime.datetime = Field(
-        default_factory=datetime.datetime.utcnow,
-        description="time of the update in UTC",
-    )
-
     @classmethod
     def from_container(cls, container: Dict[str, Any]) -> "DockerContainerInspect":
         return cls(
@@ -246,62 +239,6 @@ def are_containers_ready(self) -> bool:
         )
 
 
-class ServiceLabelsStoredData(CommonServiceDetails, DynamicSidecarServiceLabels):
-    service_name: str
-    paths_mapping: PathMappingsLabel  # overwrites in DynamicSidecarServiceLabels
-    dynamic_sidecar_network_name: str
-    simcore_traefik_zone: str
-    service_port: int
-
-    @classmethod
-    def from_service(cls, service: Dict[str, Any]) -> "ServiceLabelsStoredData":
-        labels = service["Spec"]["Labels"]
-        params = dict(
-            service_name=service["Spec"]["Name"],
-            node_uuid=NodeID(labels["uuid"]),
-            key=labels["service_key"],
-            version=labels["service_tag"],
-            paths_mapping=PathMappingsLabel.parse_raw(labels["paths_mapping"]),
-            dynamic_sidecar_network_name=labels["traefik.docker.network"],
-            simcore_traefik_zone=labels["io.simcore.zone"],
-            service_port=labels["service_port"],
-            project_id=ProjectID(labels["study_id"]),
-            user_id=int(labels["user_id"]),
-        )
-        if "compose_spec" in labels:
-            params["compose_spec"] = labels["compose_spec"]
-        if "container_http_entry" in labels:
-            params["container_http_entry"] = labels["container_http_entry"]
-        if "restart_policy" in labels:
-            params["restart_policy"] = labels["restart_policy"]
-        return cls(**params)
-
-    class Config:
-        extra = Extra.allow
-        allow_population_by_field_name = True
-        schema_extra = {
-            "example": {
-                "service_name": "some service",
-                "node_uuid": UUID("75c7f3f4-18f9-4678-8610-54a2ade78eaa"),
-                "key": "simcore/services/dynamic/3dviewer",
-                "version": "2.4.5",
-                "paths_mapping": PathMappingsLabel.parse_obj(
-                    PathMappingsLabel.Config.schema_extra["example"]
-                ),
-                "compose_spec": SimcoreServiceLabels.Config.schema_extra["examples"][2][
-                    "simcore.service.compose-spec"
-                ],
-                "container_http_entry": "some-entrypoint",
-                "dynamic_sidecar_network_name": "some_network_name",
-                "simcore_traefik_zone": "main",
-                "service_port": 300,
-                "restart_policy": RestartPolicy.NO_RESTART.value,
-                "project_id": UUID("dd1d04d9-d704-4f7e-8f0f-1ca60cc771fe"),
-                "user_id": 234,
-            }
-        }
-
-
 class DynamicSidecarNames(BaseModel):
     """
     Service naming schema:
@@ -432,38 +369,11 @@ def from_http_request(
         return cls.parse_obj(obj_dict)
 
     @classmethod
-    def from_service_labels_stored_data(
-        cls,
-        service_labels_stored_data: ServiceLabelsStoredData,
-        port: Optional[int],
-        request_dns: str = None,
-        request_scheme: str = None,
-        proxy_service_name: str = None,
+    def from_service_inspect(
+        cls, service_inspect: Mapping[str, Any]
     ) -> "SchedulerData":
-        return cls.parse_obj(
-            dict(
-                service_name=service_labels_stored_data.service_name,
-                node_uuid=service_labels_stored_data.node_uuid,
-                project_id=service_labels_stored_data.project_id,
-                user_id=service_labels_stored_data.user_id,
-                key=service_labels_stored_data.key,
-                version=service_labels_stored_data.version,
-                paths_mapping=service_labels_stored_data.paths_mapping,
-                compose_spec=json.dumps(service_labels_stored_data.compose_spec),
-                container_http_entry=service_labels_stored_data.container_http_entry,
-                restart_policy=service_labels_stored_data.restart_policy,
-                dynamic_sidecar_network_name=service_labels_stored_data.dynamic_sidecar_network_name,
-                simcore_traefik_zone=service_labels_stored_data.simcore_traefik_zone,
-                service_port=service_labels_stored_data.service_port,
-                request_dns=request_dns,
-                request_scheme=request_scheme,
-                proxy_service_name=proxy_service_name,
-                dynamic_sidecar=dict(
-                    hostname=service_labels_stored_data.service_name,
-                    port=port,
-                ),
-            )
-        )
+        labels = service_inspect["Spec"]["Labels"]
+        return cls.parse_raw(labels[DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL])
 
     class Config:
         extra = Extra.allow

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

@@ -2,27 +2,28 @@
 
 
 import asyncio
+import json
 import logging
 import time
 import traceback
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Deque, Dict, List, Optional, Set, Tuple
+from copy import deepcopy
+from typing import Any, AsyncIterator, Dict, List, Mapping, Optional, Set, Tuple
 
 import aiodocker
-from aiodocker.utils import clean_filters
+from aiodocker.utils import clean_filters, clean_map
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
 from packaging import version
 from servicelib.utils import logged_gather
 
 from ...core.settings import DynamicSidecarSettings
-from ...models.schemas.constants import DYNAMIC_SIDECAR_SERVICE_PREFIX
-from ...models.schemas.dynamic_services import (
-    ServiceLabelsStoredData,
-    ServiceState,
-    ServiceType,
+from ...models.schemas.constants import (
+    DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL,
+    DYNAMIC_SIDECAR_SERVICE_PREFIX,
 )
+from ...models.schemas.dynamic_services import SchedulerData, ServiceState, ServiceType
 from .docker_states import TASK_STATES_RUNNING, extract_task_state
 from .errors import DynamicSidecarError, GenericDockerError
 
@@ -161,24 +162,22 @@ async def inspect_service(service_id: str) -> Dict[str, Any]:
 
 async def get_dynamic_sidecars_to_observe(
     dynamic_sidecar_settings: DynamicSidecarSettings,
-) -> Deque[ServiceLabelsStoredData]:
+) -> List[SchedulerData]:
     """called when scheduler is started to discover new services to observe"""
     async with docker_client() as client:
-        running_dynamic_sidecar_services = await client.services.list(
+        running_dynamic_sidecar_services: List[
+            Mapping[str, Any]
+        ] = await client.services.list(
             filters={
                 "label": [
                     f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}"
                 ],
                 "name": [f"{DYNAMIC_SIDECAR_SERVICE_PREFIX}"],
             }
         )
-
-    dynamic_sidecar_services: Deque[ServiceLabelsStoredData] = Deque()
-
-    for service in running_dynamic_sidecar_services:
-        dynamic_sidecar_services.append(ServiceLabelsStoredData.from_service(service))
-
-    return dynamic_sidecar_services
+    return [
+        SchedulerData.from_service_inspect(x) for x in running_dynamic_sidecar_services
+    ]
 
 
 async def _extract_task_data_from_service_for_state(
@@ -374,10 +373,17 @@ async def remove_dynamic_sidecar_network(network_name: str) -> bool:
         return False
 
 
-async def remove_dynamic_sidecar_volumes(node_uuid: NodeID) -> Set[str]:
+async def remove_dynamic_sidecar_volumes(
+    node_uuid: NodeID, dynamic_sidecar_settings: DynamicSidecarSettings
+) -> Set[str]:
     async with docker_client() as client:
         volumes_response = await client.volumes.list(
-            filters={"label": f"uuid={node_uuid}"}
+            filters={
+                "label": [
+                    f"swarm_stack_name={dynamic_sidecar_settings.SWARM_STACK_NAME}",
+                    f"uuid={node_uuid}",
+                ]
+            }
         )
         volumes = volumes_response["Volumes"]
         log.debug("Removing volumes: %s", [v["Name"] for v in volumes])
@@ -506,3 +512,43 @@ async def try_to_remove_network(network_name: str) -> None:
             return await network.delete()
         except aiodocker.exceptions.DockerError:
             log.warning("Could not remove network %s", network_name)
+
+
+async def update_scheduler_data_label(scheduler_data: SchedulerData) -> None:
+    async with docker_client() as client:
+        # NOTE: builtin `DockerServices.update` function is very limited.
+        # Using the same pattern but updating labels
+
+        try:
+            # fetch information from service name
+            service_inspect = await client.services.inspect(scheduler_data.service_name)
+            service_version = service_inspect["Version"]["Index"]
+            service_id = service_inspect["ID"]
+            spec = service_inspect["Spec"]
+
+            # compose_spec needs to be json encoded
+            # before encoding it to json and storing it
+            # in the label
+            scheduler_data_copy = deepcopy(scheduler_data)
+            scheduler_data_copy.compose_spec = json.dumps(
+                scheduler_data_copy.compose_spec
+            )
+            label_data = scheduler_data_copy.json()
+            spec["Labels"][DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL] = label_data
+
+            await client._query_json(  # pylint: disable=protected-access
+                f"services/{service_id}/update",
+                method="POST",
+                data=json.dumps(clean_map(spec)),
+                params={"version": service_version},
+            )
+        except aiodocker.exceptions.DockerError as e:
+            if not (
+                e.status == 404
+                and e.message == f"service {scheduler_data.service_name} not found"
+            ):
+                raise e
+            log.debug(
+                "Skip update for service '%s' which could not be found",
+                scheduler_data.service_name,
+            )

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

@@ -5,6 +5,7 @@
 from servicelib.json_serialization import json_dumps
 
 from ....core.settings import AppSettings, DynamicSidecarSettings
+from ....models.schemas.constants import DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL
 from ....models.schemas.dynamic_services import SchedulerData, ServiceType
 from .._namepsace import get_compose_namespace
 from ..volumes_resolver import DynamicSidecarVolumesPathsResolver
@@ -102,6 +103,7 @@ def get_dynamic_sidecar_spec(
     ]:
         mounts.append(
             DynamicSidecarVolumesPathsResolver.mount_entry(
+                swarm_stack_name=dynamic_sidecar_settings.SWARM_STACK_NAME,
                 compose_namespace=compose_namespace,
                 path=path_to_mount,
                 node_uuid=scheduler_data.node_uuid,
@@ -113,6 +115,7 @@ def get_dynamic_sidecar_spec(
         if app_settings.DIRECTOR_V2_DEV_FEATURES_ENABLED:
             mounts.append(
                 DynamicSidecarVolumesPathsResolver.mount_r_clone(
+                    swarm_stack_name=dynamic_sidecar_settings.SWARM_STACK_NAME,
                     compose_namespace=compose_namespace,
                     path=path_to_mount,
                     project_id=scheduler_data.project_id,
@@ -123,6 +126,7 @@ def get_dynamic_sidecar_spec(
         else:
             mounts.append(
                 DynamicSidecarVolumesPathsResolver.mount_entry(
+                    swarm_stack_name=dynamic_sidecar_settings.SWARM_STACK_NAME,
                     compose_namespace=compose_namespace,
                     path=path_to_mount,
                     node_uuid=scheduler_data.node_uuid,
@@ -187,13 +191,8 @@ def get_dynamic_sidecar_spec(
             "user_id": f"{scheduler_data.user_id}",
             # the following are used for scheduling
             "uuid": f"{scheduler_data.node_uuid}",  # also needed for removal when project is closed
-            "swarm_stack_name": dynamic_sidecar_settings.SWARM_STACK_NAME,
-            "service_key": scheduler_data.key,
-            "service_tag": scheduler_data.version,
-            "paths_mapping": scheduler_data.paths_mapping.json(),
-            "compose_spec": json_dumps(scheduler_data.compose_spec),
-            "container_http_entry": scheduler_data.container_http_entry,
-            "restart_policy": scheduler_data.restart_policy,
+            "swarm_stack_name": dynamic_sidecar_settings.SWARM_STACK_NAME,  # required for listing services with uuid
+            DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL: scheduler_data.json(),
         },
         "name": scheduler_data.service_name,
         "networks": [swarm_network_id, dynamic_sidecar_network_id],