🐛 Avoids service shutdown due to swarm networking errors (ITISFoundation#3394)

Author: GitHK

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

@@ -0,0 +1,66 @@
+import logging
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field, NonNegativeFloat, PrivateAttr
+
+log = logging.getLogger(__name__)
+
+
+class DelayedExceptionHandler(BaseModel):
+    """
+    Allows to ignore an exception for an established
+    period of time after which it is raised.
+
+    This use case most commonly occurs when dealing with
+    external systems.
+    For example, due to poor network  performance or
+    network congestion, an external system which is healthy,
+    currently is not reachable any longer.
+    A possible solution:
+    - ignore exceptions for an interval in which the
+        system usually is reachable again by not
+        raising the error
+    - if the error persist give up and raise it
+
+    Example code usage:
+
+        delayed_handler_external_service = DelayedExceptionHandler(
+            delay_for=60
+        )
+        try:
+            function_called_periodically_accessing_external_service()
+        except TargetException as e:
+            delayed_handler_external_service.try_to_raise(e)
+        else:
+            delayed_handler_external_service.else_reset()
+    """
+
+    _first_exception_skip: Optional[datetime] = PrivateAttr(None)
+    _failure_counter: int = PrivateAttr(0)
+
+    delay_for: NonNegativeFloat = Field(
+        description="interval of time during which exceptions are ignored"
+    )
+
+    def try_to_raise(self, exception: BaseException) -> None:
+        """raises `exception` after `delay_for` passed from the first call"""
+        self._failure_counter += 1
+
+        # first time the exception was detected
+        if self._first_exception_skip is None:
+            self._first_exception_skip = datetime.utcnow()
+
+        # raise if subsequent exception is outside of delay window
+        elif (
+            datetime.utcnow() - self._first_exception_skip
+        ).total_seconds() > self.delay_for:
+            raise exception
+
+        # ignore if exception inside delay window
+        log.warning("%s skip(s) of exception: %s", self._failure_counter, exception)
+
+    def else_reset(self) -> None:
+        """error no longer occurs reset tracking"""
+        self._first_exception_skip = None
+        self._failure_counter = 0

packages/service-library/tests/test_exception_utils.py

@@ -0,0 +1,51 @@
+from math import ceil
+from time import sleep
+from typing import Final
+
+import pytest
+from pydantic import PositiveFloat, PositiveInt
+from servicelib.exception_utils import DelayedExceptionHandler
+
+TOLERANCE: Final[PositiveFloat] = 0.1
+SLEEP_FOR: Final[PositiveFloat] = TOLERANCE * 0.1
+ITERATIONS: Final[PositiveInt] = int(ceil(TOLERANCE / SLEEP_FOR)) + 1
+
+
+class TargetException(Exception):
+    pass
+
+
+def workflow(*, stop_raising_after: PositiveInt) -> int:
+    counter = 0
+
+    def function_which_can_raise():
+        nonlocal counter
+        counter += 1
+
+        if counter < stop_raising_after:
+            raise TargetException()
+
+    delayed_handler_external_service = DelayedExceptionHandler(delay_for=TOLERANCE)
+
+    def periodic_event():
+        try:
+            function_which_can_raise()
+        except TargetException as e:
+            delayed_handler_external_service.try_to_raise(e)
+        else:
+            delayed_handler_external_service.else_reset()
+
+    for _ in range(ITERATIONS):
+        periodic_event()
+        sleep(SLEEP_FOR)
+
+    return counter
+
+
+def test_workflow_passes() -> None:
+    assert workflow(stop_raising_after=2) == ITERATIONS
+
+
+def test_workflow_raises() -> None:
+    with pytest.raises(TargetException):
+        workflow(stop_raising_after=ITERATIONS + 1)

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

@@ -275,6 +275,16 @@ class DynamicSidecarSettings(BaseCustomSettings):
         ),
     )
 
+    DYNAMIC_SIDECAR_NETWORK_ISSUES_TOLERANCE_S: PositiveFloat = Field(
+        1 * MINS,
+        description=(
+            "Connectivity between director-v2 and a dy-sidecar can be "
+            "temporarily disrupted if network between swarm nodes has "
+            "issues. To avoid the sidecar being marked as failed, "
+            "allow for some time to pass before declaring it failed."
+        ),
+    )
+
     #
     # DEVELOPMENT ONLY config
     #

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

@@ -15,6 +15,7 @@
 from models_library.services_resources import ServiceResourcesDict
 from pydantic import AnyHttpUrl, BaseModel, Extra, Field, constr, parse_obj_as
 from servicelib.error_codes import ErrorCodeStr
+from servicelib.exception_utils import DelayedExceptionHandler
 
 from ..constants import (
     DYNAMIC_PROXY_SERVICE_PREFIX,
@@ -239,6 +240,15 @@ def compose_spec_submitted(self) -> bool:
         ),
     )
 
+    inspect_error_handler: DelayedExceptionHandler = Field(
+        DelayedExceptionHandler(delay_for=0),
+        description=(
+            "Set when the dy-sidecar can no longer be reached by the "
+            "director-v2. If it will be possible to reach the dy-sidecar again, "
+            "this value will be set to None."
+        ),
+    )
+
     class Config:
         validate_assignment = True
 
@@ -295,6 +305,10 @@ def make(cls, node_uuid: UUID) -> "DynamicSidecarNamesHelper":
 
 
 class SchedulerData(CommonServiceDetails, DynamicSidecarServiceLabels):
+    # TODO: ANE this object is just the context of the dy-sidecar. Should
+    # be called like so and subcontexts for different handlers should
+    # also be added. It will make keeping track of env vars more easily
+
     service_name: ServiceName = Field(
         ...,
         description="Name of the current dynamic-sidecar being observed",

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_public.py

@@ -400,10 +400,10 @@ def get_dynamic_sidecar_client(app: FastAPI) -> DynamicSidecarClient:
 
 async def get_dynamic_sidecar_service_health(
     app: FastAPI, scheduler_data: SchedulerData
-) -> None:
+) -> bool:
     api_client = get_dynamic_sidecar_client(app)
     service_endpoint = scheduler_data.endpoint
 
     # update service health
     is_healthy = await api_client.is_healthy(service_endpoint)
-    scheduler_data.dynamic_sidecar.is_available = is_healthy
+    return is_healthy

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

@@ -41,7 +41,12 @@ async def apply_observation_cycle(
             node_uuid=scheduler_data.node_uuid,
             can_save=scheduler_data.dynamic_sidecar.service_removal_state.can_save,
         )
-    await get_dynamic_sidecar_service_health(app, scheduler_data)
+
+    # TODO: ANE this can be moved to a handler in the future scheduled
+    # to all the correct cases
+    scheduler_data.dynamic_sidecar.is_available = (
+        await get_dynamic_sidecar_service_health(app, scheduler_data)
+    )
 
     for dynamic_scheduler_event in REGISTERED_EVENTS:
         if await dynamic_scheduler_event.will_trigger(

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

@@ -195,7 +195,7 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
             docker_node_id=scheduler_data.dynamic_sidecar.docker_node_id,
         )
 
-        # update service_port and assing it to the status
+        # update service_port and assign it to the status
         # needed by CreateUserServices action
         scheduler_data.service_port = extract_service_port_from_compose_start_spec(
             dynamic_sidecar_service_final_spec
@@ -230,23 +230,32 @@ async def will_trigger(cls, app: FastAPI, scheduler_data: SchedulerData) -> bool
     async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         dynamic_sidecar_client = get_dynamic_sidecar_client(app)
         dynamic_sidecar_endpoint = scheduler_data.endpoint
+        dynamic_sidecar_settings: DynamicSidecarSettings = (
+            app.state.settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR
+        )
+        scheduler_data.dynamic_sidecar.inspect_error_handler.delay_for = (
+            dynamic_sidecar_settings.DYNAMIC_SIDECAR_NETWORK_ISSUES_TOLERANCE_S
+        )
 
         try:
             containers_inspect: dict[
                 str, Any
             ] = await dynamic_sidecar_client.containers_inspect(
                 dynamic_sidecar_endpoint
             )
-        except BaseClientHTTPError:
+        except BaseClientHTTPError as e:
             were_service_containers_detected_before = (
                 len(scheduler_data.dynamic_sidecar.containers_inspect) > 0
             )
             if were_service_containers_detected_before:
                 # Containers disappeared after they were started.
                 # for now just mark as error and remove the sidecar
-                # NOTE: this is the correct place where to try and
-                # restart them (future use case).
-                raise
+
+                # NOTE: Network performance can degrade and the sidecar might
+                # be temporarily unreachable.
+                # Adding a delay between when the error is first seen and when the
+                # error is raised to avoid random shutdowns of dynamic-sidecar services.
+                scheduler_data.dynamic_sidecar.inspect_error_handler.try_to_raise(e)
 
             # After the service creation it takes a bit of time for the container to start
             # If the same message appears in the log multiple times in a row (for the same
@@ -256,12 +265,19 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
                 scheduler_data.service_name,
             )
             return
+        else:
+            scheduler_data.dynamic_sidecar.inspect_error_handler.else_reset()
 
         # parse and store data from container
         scheduler_data.dynamic_sidecar.containers_inspect = parse_containers_inspect(
             containers_inspect
         )
 
+        # TODO: ANE using `were_service_containers_detected_before` together with
+        # how many containers to expect, it can be detected if containers
+        # died and these can be restarted. Best way to go about it is
+        # to have a different handler trigger in this case registered for idling!
+
 
 class PrepareServicesEnvironment(DynamicSchedulerEvent):
     """

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

@@ -0,0 +1,116 @@
+# pylint:disable=redefined-outer-name
+# pylint:disable=unused-argument
+
+import asyncio
+from typing import Final
+
+import pytest
+from _pytest.logging import LogCaptureFixture
+from _pytest.monkeypatch import MonkeyPatch
+from fastapi import FastAPI
+from pydantic import PositiveFloat, PositiveInt
+from pytest_mock import MockerFixture
+from pytest_simcore.helpers.typing_env import EnvVarsDict
+from pytest_simcore.helpers.utils_envs import setenvs_from_dict
+from simcore_service_director_v2.models.schemas.dynamic_services import SchedulerData
+from simcore_service_director_v2.models.schemas.dynamic_services.scheduler import (
+    DockerContainerInspect,
+    DockerStatus,
+)
+from simcore_service_director_v2.modules.dynamic_sidecar.api_client import (
+    BaseClientHTTPError,
+)
+from simcore_service_director_v2.modules.dynamic_sidecar.scheduler import events
+
+NETWORK_TOLERANCE_S: Final[PositiveFloat] = 0.1
+STEPS: Final[PositiveFloat] = 10
+SLEEP_BETWEEN_CALLS: Final[PositiveFloat] = NETWORK_TOLERANCE_S / STEPS
+REPEAT_COUNT: Final[PositiveInt] = STEPS + 1
+
+
+@pytest.fixture
+def mock_env(
+    mock_env: EnvVarsDict,
+    monkeypatch: MonkeyPatch,
+) -> None:
+    setenvs_from_dict(
+        monkeypatch,
+        {
+            "S3_ENDPOINT": "",
+            "S3_ACCESS_KEY": "",
+            "S3_SECRET_KEY": "",
+            "S3_BUCKET_NAME": "",
+            "S3_SECURE": "false",
+            "POSTGRES_HOST": "",
+            "POSTGRES_USER": "",
+            "POSTGRES_PASSWORD": "",
+            "POSTGRES_DB": "",
+            "DYNAMIC_SIDECAR_NETWORK_ISSUES_TOLERANCE_S": f"{NETWORK_TOLERANCE_S}",
+        },
+    )
+
+
+@pytest.fixture
+def mock_dynamic_sidecar_client_always_fail(mocker: MockerFixture) -> None:
+    class MockedObj:
+        @classmethod
+        async def containers_inspect(cls, *args, **kwargs) -> None:
+            raise BaseClientHTTPError("will always fail")
+
+    mocker.patch.object(events, "get_dynamic_sidecar_client", return_value=MockedObj())
+
+
+@pytest.fixture
+def mock_dynamic_sidecar_client_stops_failing(mocker: MockerFixture) -> None:
+    class MockedObj:
+        def __init__(self) -> None:
+            self.counter = 0
+
+        async def containers_inspect(self, *args, **kwargs) -> None:
+            self.counter += 1
+            if self.counter < STEPS / 2:
+                raise BaseClientHTTPError("will always fail")
+
+    mocker.patch.object(events, "get_dynamic_sidecar_client", return_value=MockedObj())
+
+
+@pytest.fixture
+def docker_container_inspect() -> DockerContainerInspect:
+    return DockerContainerInspect(status=DockerStatus.DEAD, name="", id="")
+
+
+@pytest.fixture
+def scheduler_data(
+    scheduler_data: SchedulerData, docker_container_inspect: DockerContainerInspect
+) -> SchedulerData:
+    scheduler_data.dynamic_sidecar.containers_inspect = [docker_container_inspect]
+    return scheduler_data
+
+
+_CHECK_LOG_EXCEPTION_IS_SKIPPED = "skip(s) of exception"
+
+
+async def test_event_get_status_network_connectivity(
+    mock_dynamic_sidecar_client_always_fail: None,
+    minimal_app: FastAPI,
+    scheduler_data: SchedulerData,
+    caplog_info_level: LogCaptureFixture,
+):
+    with pytest.raises(BaseClientHTTPError):
+        for _ in range(REPEAT_COUNT):
+            await events.GetStatus.action(minimal_app, scheduler_data)
+            await asyncio.sleep(SLEEP_BETWEEN_CALLS)
+
+    assert caplog_info_level.text.count(_CHECK_LOG_EXCEPTION_IS_SKIPPED) > 1
+
+
+async def test_event_get_status_recovers_after_error(
+    mock_dynamic_sidecar_client_stops_failing: None,
+    minimal_app: FastAPI,
+    scheduler_data: SchedulerData,
+    caplog_info_level: LogCaptureFixture,
+):
+    for _ in range(REPEAT_COUNT):
+        await events.GetStatus.action(minimal_app, scheduler_data)
+        await asyncio.sleep(SLEEP_BETWEEN_CALLS)
+    assert caplog_info_level.text.count(_CHECK_LOG_EXCEPTION_IS_SKIPPED) >= 1