🐛♻️ Is638/fixes docker-compose operations bursts and create containers call in dy-sidecar (ITISFoundation#3187)

Author: pcrespov

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

@@ -0,0 +1,22 @@
+from typing import AsyncIterable
+
+import pytest
+from servicelib.async_utils import cancel_sequential_workers
+
+
+@pytest.fixture
+async def ensure_run_in_sequence_context_is_empty() -> AsyncIterable[None]:
+    """
+    Needed in tests calling functions decorated with 'run_sequentially_in_context'
+
+    This is a teardown only fixture
+
+    Required when shutting down the application or ending tests
+    otherwise errors will occur when closing the loop
+    """
+
+    # nothing on-startup
+
+    yield
+
+    await cancel_sequential_workers()

packages/service-library/src/servicelib/async_utils.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 from collections import deque
+from contextlib import suppress
 from dataclasses import dataclass
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Callable, Deque, Optional
@@ -32,16 +33,24 @@ class Context:
 _sequential_jobs_contexts: dict[str, Context] = {}
 
 
-async def stop_sequential_workers() -> None:
-    """Singlas all workers to close thus avoiding errors on shutdown"""
+async def cancel_sequential_workers() -> None:
+    """Signals all workers to close thus avoiding errors on shutdown"""
     for context in _sequential_jobs_contexts.values():
         await context.in_queue.put(None)
         if context.task is not None:
-            await context.task
+            context.task.cancel()
+            with suppress(asyncio.CancelledError):
+                await context.task
+
     _sequential_jobs_contexts.clear()
     logger.info("All run_sequentially_in_context pending workers stopped")
 
 
+# NOTE: If you get funny mismatches with mypy in returned values it might be due to this decorator.
+# @run_sequentially_in_contextreturn changes the return type of the decorated function to `Any`.
+# Instead we should annotate this decorator with ParamSpec and TypeVar generics.
+# SEE https://peps.python.org/pep-0612/
+#
 def run_sequentially_in_context(
     target_args: Optional[list[str]] = None,
 ) -> Callable:

packages/service-library/src/servicelib/fastapi/requests_decorators.py

@@ -38,26 +38,19 @@ def _validate_signature(handler: _HandlerWithRequestArg):
 _POLL_INTERVAL_S: float = 0.01
 
 
-async def disconnect_poller(request: Request, result: Any):
+async def _disconnect_poller(request: Request, result: Any):
     """
     Poll for a disconnect.
     If the request disconnects, stop polling and return.
     """
     while not await request.is_disconnected():
         await asyncio.sleep(_POLL_INTERVAL_S)
-
-    logger.debug(
-        "client %s disconnected! Cancelling handler for request %s %s",
-        request.client,
-        request.method,
-        request.url,
-    )
     return result
 
 
 def cancel_on_disconnect(handler: _HandlerWithRequestArg):
     """
-    After client dicsonnects, handler gets cancelled in ~<3 secs
+    After client disconnects, handler gets cancelled in ~<3 secs
     """
 
     _validate_signature(handler)
@@ -69,7 +62,7 @@ async def wrapper(request: Request, *args, **kwargs):
         # Create two tasks:
         # one to poll the request and check if the client disconnected
         poller_task = asyncio.create_task(
-            disconnect_poller(request, sentinel),
+            _disconnect_poller(request, sentinel),
             name=f"cancel_on_disconnect/poller/{handler.__name__}/{id(sentinel)}",
         )
         # , and another which is the request handler
@@ -85,15 +78,14 @@ async def wrapper(request: Request, *args, **kwargs):
         # One has completed, cancel the other
         for t in pending:
             t.cancel()
+
             try:
                 await asyncio.wait_for(t, timeout=3)
+
             except asyncio.CancelledError:
-                logger.debug("%s was cancelled", t)
-            except Exception as exc:  # pylint: disable=broad-except
+                pass
+            except Exception:  # pylint: disable=broad-except
                 if t is handler_task:
-                    logger.warning(
-                        "%s raised %s when being cancelled.", t, exc, exc_info=True
-                    )
                     raise
             finally:
                 assert t.done()  # nosec
@@ -103,15 +95,18 @@ async def wrapper(request: Request, *args, **kwargs):
             assert poller_task.done()  # nosec
             return await handler_task
 
-        # Otherwise, raise an exception. This is not exactly needed, but it will prevent
-        # validation errors if your request handler is supposed to return something.
-        logger.debug(
-            "Request %s %s cancelled:\n - %s\n - %s",
+        # Otherwise, raise an exception. This is not exactly needed,
+        # but it will prevent validation errors if your request handler
+        # is supposed to return something.
+        logger.warning(
+            "Request %s %s cancelled since client %s disconnected:\n - %s\n - %s",
             request.method,
             request.url,
+            request.client,
             f"{poller_task=}",
             f"{handler_task=}",
         )
+
         assert poller_task.done()  # nosec
         assert handler_task.done()  # nosec
 
@@ -122,3 +117,6 @@ async def wrapper(request: Request, *args, **kwargs):
         )
 
     return wrapper
+
+
+__all__: tuple[str, ...] = ("cancel_on_disconnect",)

packages/service-library/tests/conftest.py

@@ -5,15 +5,16 @@
 import sys
 from copy import deepcopy
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
 
 import pytest
 import servicelib
 from faker import Faker
 
 pytest_plugins = [
-    "pytest_simcore.repository_paths",
     "pytest_simcore.pytest_global_environs",
+    "pytest_simcore.repository_paths",
+    "pytest_simcore.simcore_service_library_fixtures",
 ]
 
 
@@ -40,7 +41,7 @@ def osparc_simcore_root_dir(here) -> Path:
 
 
 @pytest.fixture
-def fake_data_dict(faker: Faker) -> Dict[str, Any]:
+def fake_data_dict(faker: Faker) -> dict[str, Any]:
     data = {
         "uuid_as_UUID": faker.uuid4(cast_to=None),
         "uuid_as_str": faker.uuid4(),

packages/service-library/tests/test_async_utils.py

@@ -8,29 +8,19 @@
 from collections import deque
 from dataclasses import dataclass
 from time import time
-from typing import Any, AsyncIterable, Optional
+from typing import Any, Optional
 
 import pytest
 from faker import Faker
 from servicelib.async_utils import (
     _sequential_jobs_contexts,
     run_sequentially_in_context,
-    stop_sequential_workers,
 )
 
 RETRIES = 10
 DIFFERENT_CONTEXTS_COUNT = 10
 
 
-@pytest.fixture
-async def ensure_run_in_sequence_context_is_empty() -> AsyncIterable[None]:
-    yield
-    # NOTE
-    # required when shutting down the application or ending tests
-    # otherwise errors will occur when closing the loop
-    await stop_sequential_workers()
-
-
 @pytest.fixture
 def payload(faker: Faker) -> str:
     return faker.text()

services/director-v2/src/simcore_service_director_v2/api/routes/dynamic_services.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import logging
 from typing import Coroutine, Optional, Union, cast
 from uuid import UUID
@@ -12,6 +11,7 @@
 from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services import ServiceKeyVersion
 from models_library.users import UserID
+from servicelib.json_serialization import json_dumps
 from starlette import status
 from starlette.datastructures import URL
 from tenacity import RetryCallState, TryAgain
@@ -88,6 +88,8 @@ async def list_running_dynamic_services(
             dynamic_services_settings.DYNAMIC_SIDECAR, user_id, project_id
         )
     ]
+
+    # NOTE: Review error handling https://github.com/ITISFoundation/osparc-simcore/issues/3194
     dynamic_sidecar_running_services: list[DynamicServiceOut] = cast(
         list[DynamicServiceOut], await asyncio.gather(*get_stack_statuse_tasks)
     )
@@ -214,7 +216,7 @@ def _log_error(retry_state: RetryCallState):
         logger.error(
             "Service with %s could not be untracked after %s",
             f"{node_uuid=}",
-            f"{json.dumps(retry_state.retry_object.statistics)}",
+            f"{json_dumps(retry_state.retry_object.statistics)}",
         )
 
     async for attempt in AsyncRetrying(

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

@@ -1,4 +1,4 @@
-from servicelib.async_utils import stop_sequential_workers
+from servicelib.async_utils import cancel_sequential_workers
 
 from ..meta import PROJECT_NAME, __version__
 
@@ -11,7 +11,7 @@
 | | | |_ _ __ ___  ___| |_ ___  _ __
 | | | | | '__/ _ \/ __| __/ _ \| '__|
 | |/ /| | | |  __/ (__| || (_) | |
-|___/ |_|_|  \___|\___|\__\___/|_|   {0}
+|___/ |_|_|  \___|\___|\__\___/|_|   {}
 
 """.format(
     f"v{__version__}"
@@ -23,6 +23,6 @@ async def on_startup() -> None:
 
 
 async def on_shutdown() -> None:
-    await stop_sequential_workers()
+    await cancel_sequential_workers()
     msg = PROJECT_NAME + f" v{__version__} SHUT DOWN"
     print(f"{msg:=^100}", flush=True)

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

@@ -6,7 +6,7 @@
 import logging
 import time
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Mapping, Optional
+from typing import Any, AsyncIterator, Mapping, Optional, Union
 
 import aiodocker
 from aiodocker.utils import clean_filters, clean_map
@@ -16,6 +16,7 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
+from servicelib.json_serialization import json_dumps
 from servicelib.utils import logged_gather
 from tenacity._asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
@@ -106,10 +107,21 @@ async def create_network(network_config: dict[str, Any]) -> str:
             )
 
 
-async def create_service_and_get_id(create_service_data: AioDockerServiceSpec) -> str:
+async def create_service_and_get_id(
+    create_service_data: Union[AioDockerServiceSpec, dict[str, Any]]
+) -> str:
+    # NOTE: ideally the argument should always be AioDockerServiceSpec
+    # but for that we need get_dynamic_proxy_spec to return that type
     async with docker_client() as client:
-        service_start_result = await client.services.create(
-            **jsonable_encoder(create_service_data, by_alias=True, exclude_unset=True)
+        kwargs = jsonable_encoder(
+            create_service_data, by_alias=True, exclude_unset=True
+        )
+        service_start_result = await client.services.create(**kwargs)
+
+        log.debug(
+            "Started service %s with\n%s",
+            service_start_result,
+            json.dumps(kwargs, indent=1),
         )
 
     if "ID" not in service_start_result:
@@ -489,7 +501,7 @@ async def _update_service_spec(
                     await client._query_json(  # pylint: disable=protected-access
                         f"services/{service_id}/update",
                         method="POST",
-                        data=json.dumps(clean_map(updated_spec)),
+                        data=json_dumps(clean_map(updated_spec)),
                         params={"version": service_version},
                     )
                 except aiodocker.exceptions.DockerError as e:

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

@@ -1,7 +1,6 @@
-import json
 import logging
 from copy import deepcopy
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
 from fastapi.applications import FastAPI
 from models_library.service_settings_labels import ComposeSpecLabel, PathMappingsLabel
@@ -12,10 +11,11 @@
     ServiceResourcesDict,
 )
 from servicelib.docker_compose import replace_env_vars_in_compose_spec
+from servicelib.json_serialization import json_dumps
 from settings_library.docker_registry import RegistrySettings
 
-EnvKeyEqValueList = List[str]
-EnvVarsMap = Dict[str, Optional[str]]
+EnvKeyEqValueList = list[str]
+EnvVarsMap = dict[str, Optional[str]]
 
 
 logger = logging.getLogger(__name__)
@@ -97,7 +97,7 @@ def _update_paths_mappings(
         env_vars["DY_SIDECAR_PATH_OUTPUTS"] = f"{path_mappings.outputs_path}"
         env_vars[
             "DY_SIDECAR_STATE_PATHS"
-        ] = f"{json.dumps([f'{p}' for p in path_mappings.state_paths])}"
+        ] = f"{json_dumps( { f'{p}' for p in path_mappings.state_paths } )}"
 
         service_content["environment"] = _environment_section.export_as_list(env_vars)