Merge branch 'master' into pr-osparc-debug-web-worker

Author: GitHK

packages/pytest-simcore/src/pytest_simcore/helpers/docker.py

@@ -37,7 +37,7 @@ class ContainerStatus(str, Enum):
 
 _COLOR_ENCODING_RE = re.compile(r"\x1B\[([0-9]{1,2}(;[0-9]{1,2})?)?[mGK]")
 _MAX_PATH_CHAR_LEN_ALLOWED = 260
-_kFILENAME_TOO_LONG = 36
+_FILENAME_TOO_LONG = 36
 _NORMPATH_COUNT = 0
 
 
@@ -94,7 +94,7 @@ def get_service_published_port(
         )
 
         for target_port in ports_to_look_for:
-            target_port = int(target_port)
+            target_port = int(target_port)  # noqa: PLW2901
             for p in service_ports:
                 if p["TargetPort"] == target_port:
                     published_port = p["PublishedPort"]
@@ -158,7 +158,7 @@ def run_docker_compose_config(
     args = [f"{docker_compose_path}", *bash_options]
     print(" ".join(args))
 
-    process = subprocess.run(
+    process = subprocess.run(  # noqa: S603
         args,
         cwd=project_dir,
         capture_output=True,
@@ -189,7 +189,7 @@ def shorten_path(filename: str) -> Path:
     # This helper function tries to normalize the path
     # Another possibility would be that the path has some
     # problematic characters but so far we did not find any case ...
-    global _NORMPATH_COUNT  # pylint: disable=global-statement
+    global _NORMPATH_COUNT  # pylint: disable=global-statement  # noqa: PLW0603
 
     if len(filename) > _MAX_PATH_CHAR_LEN_ALLOWED:
         _NORMPATH_COUNT += 1
@@ -215,7 +215,7 @@ def safe_artifact_name(name: str) -> str:
     return BANNED_CHARS_FOR_ARTIFACTS.sub("_", name)
 
 
-def save_docker_infos(destination_dir: Path):
+def save_docker_infos(destination_dir: Path):  # noqa: C901
     client = docker.from_env()
 
     # Includes stop containers, which might be e.g. failing tasks
@@ -228,7 +228,7 @@ def save_docker_infos(destination_dir: Path):
             destination_dir.mkdir(parents=True, exist_ok=True)
 
         except OSError as err:
-            if err.errno == _kFILENAME_TOO_LONG:
+            if err.errno == _FILENAME_TOO_LONG:
                 destination_dir = shorten_path(err.filename)
                 destination_dir.mkdir(parents=True, exist_ok=True)
 
@@ -245,7 +245,7 @@ def save_docker_infos(destination_dir: Path):
                     )
 
                 except OSError as err:
-                    if err.errno == _kFILENAME_TOO_LONG:
+                    if err.errno == _FILENAME_TOO_LONG:
                         shorten_path(err.filename).write_text(
                             _COLOR_ENCODING_RE.sub("", logs)
                         )
@@ -256,12 +256,12 @@ def save_docker_infos(destination_dir: Path):
                         json.dumps(container.attrs, indent=2)
                     )
                 except OSError as err:
-                    if err.errno == _kFILENAME_TOO_LONG:
+                    if err.errno == _FILENAME_TOO_LONG:
                         shorten_path(err.filename).write_text(
                             json.dumps(container.attrs, indent=2)
                         )
 
-            except Exception as err:  # pylint: disable=broad-except  # noqa: PERF203
+            except Exception as err:  # pylint: disable=broad-except
                 if container.status != ContainerStatus.created:
                     print(
                         f"Error while dumping {container.name=}, {container.status=}.\n\t{err=}"

packages/pytest-simcore/src/pytest_simcore/helpers/paused_container.py

@@ -0,0 +1,64 @@
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import TYPE_CHECKING, Protocol
+
+from tenacity.asyncio import AsyncRetrying
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_fixed
+
+if TYPE_CHECKING:
+    from servicelib.rabbitmq import RabbitMQClient
+    from servicelib.redis import RedisClientSDK
+
+
+class _ClientWithPingProtocol(Protocol):
+    async def ping(self) -> bool: ...
+
+
+@asynccontextmanager
+async def _paused_container(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    container_name: str,
+    client: _ClientWithPingProtocol,
+) -> AsyncIterator[None]:
+    async with paused_container(container_name):
+        yield
+
+    async for attempt in AsyncRetrying(
+        wait=wait_fixed(0.1),
+        stop=stop_after_delay(10),
+        reraise=True,
+        retry=retry_if_exception_type(AssertionError),
+    ):
+        with attempt:
+            assert await client.ping() is True
+
+
+@asynccontextmanager
+async def pause_rabbit(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    rabbit_client: "RabbitMQClient",
+) -> AsyncIterator[None]:
+    """
+    Pause RabbitMQ container during the context block,
+    ensuring it's fully down before and back up after.
+    """
+    async with _paused_container(paused_container, "rabbit", rabbit_client):
+        yield
+
+
+@asynccontextmanager
+async def pause_redis(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    redis_client: "RedisClientSDK",
+) -> AsyncIterator[None]:
+    """
+    Pause Redis container during the context block,
+    saving a DB snapshot first for a clean restore point.
+    Ensures Redis is down before yielding, and back up after.
+    """
+    await redis_client.redis.save()
+
+    async with _paused_container(paused_container, "redis", redis_client):
+        yield

packages/service-library/tests/deferred_tasks/test_deferred_tasks.py

@@ -2,22 +2,24 @@
 # pylint:disable=unused-argument
 
 import asyncio
-import contextlib
+import datetime
 import itertools
 import json
 import random
 import sys
-from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterable, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, suppress
 from pathlib import Path
-from typing import Any, Protocol
+from typing import Any
 
 import psutil
 import pytest
 from aiohttp.test_utils import unused_port
 from common_library.json_serialization import json_dumps
 from common_library.serialization import model_dump_with_secrets
 from pydantic import NonNegativeFloat, NonNegativeInt
+from pytest_mock import MockerFixture
+from pytest_simcore.helpers.paused_container import pause_rabbit, pause_redis
 from servicelib.rabbitmq import RabbitMQClient
 from servicelib.redis import RedisClientSDK
 from servicelib.sequences_utils import partition_gen
@@ -330,57 +332,12 @@ async def rabbit_client(
     return create_rabbitmq_client("pinger")
 
 
-class ClientWithPingProtocol(Protocol):
-    async def ping(self) -> bool: ...
-
-
-class ServiceManager:
-    def __init__(
-        self,
-        redis_client: RedisClientSDK,
-        rabbit_client: RabbitMQClient,
-        paused_container: Callable[[str], AbstractAsyncContextManager[None]],
-    ) -> None:
-        self.redis_client = redis_client
-        self.rabbit_client = rabbit_client
-        self.paused_container = paused_container
-
-    @contextlib.asynccontextmanager
-    async def _paused_container(
-        self, container_name: str, client: ClientWithPingProtocol
-    ) -> AsyncIterator[None]:
-        async with self.paused_container(container_name):
-            async for attempt in AsyncRetrying(
-                wait=wait_fixed(0.1),
-                stop=stop_after_delay(10),
-                reraise=True,
-                retry=retry_if_exception_type(AssertionError),
-            ):
-                with attempt:
-                    assert await client.ping() is False
-            yield
-
-        async for attempt in AsyncRetrying(
-            wait=wait_fixed(0.1),
-            stop=stop_after_delay(10),
-            reraise=True,
-            retry=retry_if_exception_type(AssertionError),
-        ):
-            with attempt:
-                assert await client.ping() is True
-
-    @contextlib.asynccontextmanager
-    async def pause_rabbit(self) -> AsyncIterator[None]:
-        async with self._paused_container("rabbit", self.rabbit_client):
-            yield
-
-    @contextlib.asynccontextmanager
-    async def pause_redis(self) -> AsyncIterator[None]:
-        # save db for clean restore point
-        await self.redis_client.redis.save()
-
-        async with self._paused_container("redis", self.redis_client):
-            yield
+@pytest.fixture
+def mock_default_socket_timeout(mocker: MockerFixture) -> None:
+    mocker.patch(
+        "servicelib.redis._client.DEFAULT_SOCKET_TIMEOUT",
+        datetime.timedelta(seconds=0.25),
+    )
 
 
 @pytest.mark.parametrize("max_workers", [10])
@@ -397,9 +354,6 @@ async def test_workflow_with_third_party_services_outages(
     deferred_tasks_to_start: int,
     service: str,
 ):
-    service_manager = ServiceManager(
-        redis_client_sdk_deferred_tasks, rabbit_client, paused_container
-    )
 
     async with _RemoteProcessLifecycleManager(
         await get_remote_process(),
@@ -423,14 +377,16 @@ async def test_workflow_with_third_party_services_outages(
         match service:
             case "rabbit":
                 print("[rabbit]: pausing")
-                async with service_manager.pause_rabbit():
+                async with pause_rabbit(paused_container, rabbit_client):
                     print("[rabbit]: paused")
                     await _sleep_in_interval(0.2, 0.4)
                 print("[rabbit]: resumed")
 
             case "redis":
                 print("[redis]: pausing")
-                async with service_manager.pause_redis():
+                async with pause_redis(
+                    paused_container, redis_client_sdk_deferred_tasks
+                ):
                     print("[redis]: paused")
                     await _sleep_in_interval(0.2, 0.4)
                 print("[redis]: resumed")

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/rest/_ops.py

@@ -5,7 +5,7 @@
     DynamicServiceGet,
 )
 
-from ...services import scheduler_interface
+from ...services import common_interface
 from ._dependencies import (
     get_app,
 )
@@ -19,6 +19,6 @@ async def running_services(
 ) -> list[DynamicServiceGet]:
     """returns all running dynamic services. Used by ops internall to determine
     when it is safe to shutdown the platform"""
-    return await scheduler_interface.list_tracked_dynamic_services(
+    return await common_interface.list_tracked_dynamic_services(
         app, user_id=None, project_id=None
     )

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/rpc/_services.py

@@ -20,7 +20,7 @@
     ServiceWasNotFoundError,
 )
 
-from ...services import scheduler_interface
+from ...services import common_interface
 
 router = RPCRouter()
 
@@ -29,7 +29,7 @@
 async def list_tracked_dynamic_services(
     app: FastAPI, *, user_id: UserID | None = None, project_id: ProjectID | None = None
 ) -> list[DynamicServiceGet]:
-    return await scheduler_interface.list_tracked_dynamic_services(
+    return await common_interface.list_tracked_dynamic_services(
         app, user_id=user_id, project_id=project_id
     )
 
@@ -38,14 +38,14 @@ async def list_tracked_dynamic_services(
 async def get_service_status(
     app: FastAPI, *, node_id: NodeID
 ) -> NodeGet | DynamicServiceGet | NodeGetIdle:
-    return await scheduler_interface.get_service_status(app, node_id=node_id)
+    return await common_interface.get_service_status(app, node_id=node_id)
 
 
 @router.expose()
 async def run_dynamic_service(
     app: FastAPI, *, dynamic_service_start: DynamicServiceStart
 ) -> NodeGet | DynamicServiceGet:
-    return await scheduler_interface.run_dynamic_service(
+    return await common_interface.run_dynamic_service(
         app, dynamic_service_start=dynamic_service_start
     )
 
@@ -59,7 +59,7 @@ async def run_dynamic_service(
 async def stop_dynamic_service(
     app: FastAPI, *, dynamic_service_stop: DynamicServiceStop
 ) -> None:
-    return await scheduler_interface.stop_dynamic_service(
+    return await common_interface.stop_dynamic_service(
         app, dynamic_service_stop=dynamic_service_stop
     )
 
@@ -68,25 +68,25 @@ async def stop_dynamic_service(
 async def get_project_inactivity(
     app: FastAPI, *, project_id: ProjectID, max_inactivity_seconds: NonNegativeInt
 ) -> GetProjectInactivityResponse:
-    return await scheduler_interface.get_project_inactivity(
+    return await common_interface.get_project_inactivity(
         app, project_id=project_id, max_inactivity_seconds=max_inactivity_seconds
     )
 
 
 @router.expose()
 async def restart_user_services(app: FastAPI, *, node_id: NodeID) -> None:
-    await scheduler_interface.restart_user_services(app, node_id=node_id)
+    await common_interface.restart_user_services(app, node_id=node_id)
 
 
 @router.expose()
 async def retrieve_inputs(
     app: FastAPI, *, node_id: NodeID, port_keys: list[ServicePortKey]
 ) -> RetrieveDataOutEnveloped:
-    return await scheduler_interface.retrieve_inputs(
+    return await common_interface.retrieve_inputs(
         app, node_id=node_id, port_keys=port_keys
     )
 
 
 @router.expose()
 async def update_projects_networks(app: FastAPI, *, project_id: ProjectID) -> None:
-    await scheduler_interface.update_projects_networks(app, project_id=project_id)
+    await common_interface.update_projects_networks(app, project_id=project_id)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/events.py

@@ -23,6 +23,7 @@
 from ..services.deferred_manager import deferred_manager_lifespan
 from ..services.director_v0 import director_v0_lifespan
 from ..services.director_v2 import director_v2_lifespan
+from ..services.generic_scheduler import generic_scheduler_lifespan
 from ..services.notifier import get_notifier_lifespans
 from ..services.rabbitmq import rabbitmq_lifespan
 from ..services.redis import redis_lifespan
@@ -79,6 +80,8 @@ def create_app_lifespan(
     for lifespan in get_notifier_lifespans():
         app_lifespan.add(lifespan)
 
+    app_lifespan.add(generic_scheduler_lifespan)
+
     app_lifespan.add(service_tracker_lifespan)
     app_lifespan.add(deferred_manager_lifespan)
     app_lifespan.add(status_monitor_lifespan)