🐛dynamic-sidecar fixes (ITISFoundation#2990)

* fix issues with service update

* fix description

* fix pylint

* catching failed failed block stopping monitor

* fix pylint

* using same encoding pattern

* pylint

* capture correct error

* using explicit status

* update comment

* migrating to pydantic

* fix codestyle

* bumping down attempts to 3

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

Author: GitHK

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

@@ -375,6 +375,13 @@ def from_service_inspect(
         labels = service_inspect["Spec"]["Labels"]
         return cls.parse_raw(labels[DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL])
 
+    def as_label_data(self) -> str:
+        # compose_spec needs to be json encoded before encoding it to json
+        # and storing it in the label
+        return self.copy(
+            update={"compose_spec": json.dumps(self.compose_spec)}, deep=True
+        ).json()
+
     class Config:
         extra = Extra.allow
         allow_population_by_field_name = True

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

@@ -7,16 +7,20 @@
 import time
 import traceback
 from contextlib import asynccontextmanager
-from copy import deepcopy
 from typing import Any, AsyncIterator, Dict, List, Mapping, Optional, Set, Tuple
 
 import aiodocker
 from aiodocker.utils import clean_filters, clean_map
+from fastapi import status
 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 tenacity._asyncio import AsyncRetrying
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_attempt
+from tenacity.wait import wait_exponential
 
 from ...core.settings import DynamicSidecarSettings
 from ...models.schemas.constants import (
@@ -79,6 +83,10 @@ async def _custom_volumes_get(self, id):  # pylint: disable=redefined-builtin
 _monkey_patch_aiodocker()
 
 
+class _RetryError(Exception):
+    pass
+
+
 @asynccontextmanager
 async def docker_client() -> AsyncIterator[aiodocker.docker.Docker]:
     client = None
@@ -519,36 +527,50 @@ async def update_scheduler_data_label(scheduler_data: SchedulerData) -> None:
         # 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,
-            )
+        # The docker service update API is async, so `update out of sequence` error
+        # might get raised. This is caused by the `service_version` being out of sync
+        # with what is currently stored in the docker daemon.
+        async for attempt in AsyncRetrying(
+            # waits 1, 4, 8 seconds between retries and gives up
+            stop=stop_after_attempt(3),
+            wait=wait_exponential(),
+            retry=retry_if_exception_type(_RetryError),
+            reraise=True,
+        ):
+            with attempt:
+                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"]
+
+                    spec["Labels"][
+                        DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL
+                    ] = scheduler_data.as_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 == status.HTTP_404_NOT_FOUND
+                        and e.message
+                        == f"service {scheduler_data.service_name} not found"
+                    ):
+                        raise e
+                    if (
+                        e.status == status.HTTP_500_INTERNAL_SERVER_ERROR
+                        and e.message
+                        == "rpc error: code = Unknown desc = update out of sequence"
+                    ):
+                        raise _RetryError() from 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

@@ -192,7 +192,7 @@ def get_dynamic_sidecar_spec(
             # 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,  # required for listing services with uuid
-            DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL: scheduler_data.json(),
+            DYNAMIC_SIDECAR_SCHEDULER_DATA_LABEL: scheduler_data.as_label_data(),
         },
         "name": scheduler_data.service_name,
         "networks": [swarm_network_id, dynamic_sidecar_network_id],

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

@@ -40,7 +40,11 @@
     update_scheduler_data_label,
 )
 from ..docker_states import ServiceState, extract_containers_minimim_statuses
-from ..errors import DynamicSidecarError, DynamicSidecarNotFoundError
+from ..errors import (
+    DynamicSidecarError,
+    DynamicSidecarNotFoundError,
+    GenericDockerError,
+)
 from .events import REGISTERED_EVENTS
 
 logger = logging.getLogger(__name__)
@@ -368,7 +372,12 @@ async def observing_single_service(service_name: str) -> None:
             finally:
                 # when done, always unlock the resource
                 if scheduler_data_copy != scheduler_data:
-                    await update_scheduler_data_label(scheduler_data)
+                    try:
+                        await update_scheduler_data_label(scheduler_data)
+                    except GenericDockerError as e:
+                        logger.warning(
+                            "Skipped labels update, please check:\n %s", f"{e}"
+                        )
                 await lock_with_scheduler_data.resource_lock.unlock_resource()
 
         service_name: Optional[str]

services/director-v2/tests/unit/test_models_schemas_dynamic_services_scheduler.py

@@ -0,0 +1,18 @@
+import json
+from copy import deepcopy
+
+from simcore_service_director_v2.models.schemas.dynamic_services import SchedulerData
+
+
+def test_regression_as_label_data(scheduler_data: SchedulerData) -> None:
+    # old tested implementation
+    scheduler_data_copy = deepcopy(scheduler_data)
+    scheduler_data_copy.compose_spec = json.dumps(scheduler_data_copy.compose_spec)
+    json_encoded = scheduler_data_copy.json()
+
+    # using pydantic's internals
+    label_data = scheduler_data.as_label_data()
+
+    parsed_json_encoded = SchedulerData.parse_raw(json_encoded)
+    parsed_label_data = SchedulerData.parse_raw(label_data)
+    assert parsed_json_encoded == parsed_label_data