♻️ observation cycle is skipped when error detected (ITISFoundation#3195)

* observation cycle is skipped when error detected

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

Author: 3 people

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

@@ -353,8 +353,21 @@ def _enqueue_observation_from_service_name(self, service_name: str) -> None:
     async def _run_trigger_observation_queue_task(self) -> None:
         """generates events at regular time interval"""
 
-        async def observing_single_service(service_name: str) -> None:
+        async def _observing_single_service(service_name: str) -> None:
             scheduler_data: SchedulerData = self._to_observe[service_name]
+
+            # NOTE: ANE: not very readable should be refactored
+            if (
+                scheduler_data.dynamic_sidecar.status.current
+                == DynamicSidecarStatus.FAILING
+            ):
+                # Nothing will be done if there is an error while interacting
+                # with the sidecar.
+                # It makes no sense to continuously occupy resources or create
+                # issues due to high request to components like the `docker damon`
+                # and the `storage service`.
+                return
+
             scheduler_data_copy: SchedulerData = deepcopy(scheduler_data)
             try:
                 await _apply_observation_cycle(self.app, self, scheduler_data)
@@ -402,7 +415,7 @@ async def observing_single_service(service_name: str) -> None:
                 self._service_observation_task[
                     service_name
                 ] = observation_task = asyncio.create_task(
-                    observing_single_service(service_name),
+                    _observing_single_service(service_name),
                     name=f"observe_{service_name}",
                 )
                 observation_task.add_done_callback(

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

@@ -32,6 +32,7 @@
 from models_library.services import ServiceKeyVersion
 from pydantic.types import NonNegativeInt
 from pytest import MonkeyPatch
+from pytest_mock.plugin import MockerFixture
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from settings_library.s3 import S3Settings
 from simcore_sdk.node_ports_v2 import FileLinkType
@@ -45,6 +46,7 @@
 from simcore_service_director_v2.models.schemas.dynamic_services import (
     SchedulerData,
     ServiceDetails,
+    ServiceState,
 )
 from yarl import URL
 
@@ -379,3 +381,21 @@ def mocked_catalog_service_api(
 def caplog_info_level(caplog: LogCaptureFixture) -> Iterable[LogCaptureFixture]:
     with caplog.at_level(logging.INFO):
         yield caplog
+
+
+@pytest.fixture
+def mock_docker_api(mocker: MockerFixture) -> None:
+    mocker.patch(
+        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.get_dynamic_sidecars_to_observe",
+        autospec=True,
+        return_value=[],
+    )
+    mocker.patch(
+        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.are_all_services_present",
+        autospec=True,
+        return_value=True,
+    )
+    mocker.patch(
+        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.get_dynamic_sidecar_state",
+        return_value=(ServiceState.PENDING, ""),
+    )

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

@@ -203,24 +203,6 @@ def mocked_api_client(scheduler_data: SchedulerData) -> Iterator[MockRouter]:
         yield mock
 
 
-@pytest.fixture
-def mock_docker_api(mocker: MockerFixture) -> None:
-    mocker.patch(
-        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.get_dynamic_sidecars_to_observe",
-        autospec=True,
-        return_value=[],
-    )
-    mocker.patch(
-        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.are_all_services_present",
-        autospec=True,
-        return_value=True,
-    )
-    mocker.patch(
-        "simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task.get_dynamic_sidecar_state",
-        return_value=(ServiceState.PENDING, ""),
-    )
-
-
 @pytest.fixture
 def mock_service_running(mock_docker_api, mocker: MockerFixture) -> Iterator[AsyncMock]:
     mock = mocker.patch(

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

@@ -0,0 +1,145 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+
+import asyncio
+import re
+from typing import Final, Iterator
+
+import httpx
+import pytest
+import respx
+from _pytest.monkeypatch import MonkeyPatch
+from fastapi import FastAPI
+from pytest_simcore.helpers.typing_env import EnvVarsDict
+from pytest_simcore.helpers.utils_envs import setenvs_from_dict
+from respx.router import MockRouter
+from simcore_service_director_v2.models.schemas.dynamic_services.scheduler import (
+    SchedulerData,
+)
+from simcore_service_director_v2.modules.dynamic_sidecar.scheduler.events import (
+    REGISTERED_EVENTS,
+    DynamicSchedulerEvent,
+)
+from simcore_service_director_v2.modules.dynamic_sidecar.scheduler.task import (
+    DynamicSidecarsScheduler,
+)
+
+SCHEDULER_INTERVAL_SECONDS: Final[float] = 0.1
+
+# FIXTURES
+
+
+@pytest.fixture
+def mock_env(
+    mock_env: EnvVarsDict,
+    monkeypatch: MonkeyPatch,
+    simcore_services_network_name: str,
+    docker_swarm: None,
+    mock_docker_api: None,
+) -> None:
+    disabled_services_envs = {
+        "S3_ENDPOINT": "",
+        "S3_ACCESS_KEY": "",
+        "S3_SECRET_KEY": "",
+        "S3_BUCKET_NAME": "",
+        "POSTGRES_HOST": "",
+        "POSTGRES_USER": "",
+        "POSTGRES_PASSWORD": "",
+        "POSTGRES_DB": "",
+    }
+    setenvs_from_dict(monkeypatch, disabled_services_envs)
+
+    monkeypatch.setenv("DIRECTOR_V2_DYNAMIC_SCHEDULER_ENABLED", "true")
+    monkeypatch.setenv(
+        "DIRECTOR_V2_DYNAMIC_SCHEDULER_INTERVAL_SECONDS",
+        f"{SCHEDULER_INTERVAL_SECONDS}",
+    )
+
+
+@pytest.fixture
+def scheduler_data(scheduler_data_from_http_request: SchedulerData) -> SchedulerData:
+    return scheduler_data_from_http_request
+
+
+@pytest.fixture
+def mock_containers_docker_status(
+    scheduler_data: SchedulerData,
+) -> Iterator[MockRouter]:
+    service_endpoint = scheduler_data.dynamic_sidecar.endpoint
+    with respx.mock as mock:
+        mock.get(
+            re.compile(
+                rf"^http://{scheduler_data.service_name}:{scheduler_data.dynamic_sidecar.port}/v1/containers\?only_status=true"
+            ),
+            name="containers_docker_status",
+        ).mock(httpx.Response(200, json={}))
+        mock.get(f"{service_endpoint}/health", name="is_healthy").respond(
+            json=dict(is_healthy=True)
+        )
+
+        yield mock
+
+
+@pytest.fixture
+def scheduler(
+    mock_containers_docker_status: MockRouter, minimal_app: FastAPI
+) -> DynamicSidecarsScheduler:
+    return minimal_app.state.dynamic_sidecar_scheduler
+
+
+class ACounter:
+    def __init__(self, start: int = 0) -> None:
+        self.start = start
+        self.count = start
+
+    def increment(self) -> None:
+        self.count += 1
+
+
+@pytest.fixture
+def mocked_dynamic_scheduler_events() -> ACounter:
+    counter = ACounter()
+
+    class AlwaysTriggersDynamicSchedulerEvent(DynamicSchedulerEvent):
+        @classmethod
+        async def will_trigger(
+            cls, app: FastAPI, scheduler_data: SchedulerData
+        ) -> bool:
+            return True
+
+        @classmethod
+        async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
+            counter.increment()
+            raise RuntimeError("Failed as planned")
+
+    test_defined_scheduler_events: list[type[DynamicSchedulerEvent]] = [
+        AlwaysTriggersDynamicSchedulerEvent
+    ]
+
+    # replace REGISTERED EVENTS
+    REGISTERED_EVENTS.clear()
+    for event in test_defined_scheduler_events:
+        REGISTERED_EVENTS.append(event)
+
+    return counter
+
+
+# TESTS
+
+
+async def test_skip_observation_cycle_after_error(
+    minimal_app: FastAPI,
+    scheduler: DynamicSidecarsScheduler,
+    scheduler_data: SchedulerData,
+    mocked_dynamic_scheduler_events: ACounter,
+):
+    # add a task, emulate an error make sure no observation cycle is
+    # being triggered again
+    assert mocked_dynamic_scheduler_events.count == 0
+    await scheduler.add_service(scheduler_data)
+
+    # ensure observation cycle triggers a lot
+    await asyncio.sleep(SCHEDULER_INTERVAL_SECONDS * 10)
+    # only expect the event to be triggered once, when it raised
+    # an error and no longer trigger again
+    assert mocked_dynamic_scheduler_events.count == 1