♻️ Refactor dy-sidecar output directories event detection (ITISFoundation#3476)

Author: GitHK

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

@@ -2,7 +2,7 @@
 import contextlib
 import datetime
 import logging
-from typing import Awaitable, Callable
+from typing import Awaitable, Callable, Optional
 
 from servicelib.logging_utils import log_catch, log_context
 from tenacity import TryAgain
@@ -29,7 +29,7 @@ async def _periodic_scheduled_task(
             ), log_catch(logger):
                 await task(**task_kwargs)
 
-            raise TryAgain
+            raise TryAgain()
 
 
 async def start_periodic_task(
@@ -53,11 +53,14 @@ async def start_periodic_task(
         )
 
 
-async def stop_periodic_task(asyncio_task: asyncio.Task) -> None:
+async def stop_periodic_task(
+    asyncio_task: asyncio.Task, *, timeout: Optional[float] = None
+) -> None:
     with log_context(
         logger,
         logging.INFO,
         msg=f"cancel periodic background task '{asyncio_task.get_name()}'",
     ), contextlib.suppress(asyncio.CancelledError):
         asyncio_task.cancel()
-        await asyncio_task
+        with log_catch(logger, reraise=False):
+            await asyncio.wait((asyncio_task,), timeout=timeout)

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

@@ -3,7 +3,6 @@
 _DOCKER_TIMESTAMP_LENGTH = len("2020-10-09T12:28:14.771034")
 
 
-
 def to_datetime(docker_timestamp: str) -> datetime:
     # datetime_str is typically '2020-10-09T12:28:14.771034099Z'
     #  - The T separates the date portion from the time-of-day portion

packages/service-library/tests/test_background_task.py

@@ -6,11 +6,12 @@
 
 import asyncio
 import datetime
-from typing import AsyncIterator, Awaitable, Callable
+from typing import AsyncIterator, Awaitable, Callable, Optional
 from unittest import mock
 
 import pytest
 from faker import Faker
+from pytest import FixtureRequest
 from pytest_mock.plugin import MockerFixture
 from servicelib.background_task import start_periodic_task, stop_periodic_task
 
@@ -28,9 +29,14 @@ def task_interval() -> datetime.timedelta:
     return datetime.timedelta(seconds=_FAST_POLL_INTERVAL)
 
 
+@pytest.fixture(params=[None, 1])
+def stop_task_timeout(request: FixtureRequest) -> Optional[float]:
+    return request.param
+
+
 @pytest.fixture
 async def create_background_task(
-    faker: Faker,
+    faker: Faker, stop_task_timeout: Optional[float]
 ) -> AsyncIterator[Callable[[datetime.timedelta, Callable], Awaitable[asyncio.Task]]]:
     created_tasks = []
 
@@ -48,7 +54,9 @@ async def _creator(
 
     yield _creator
     # cleanup
-    await asyncio.gather(*(stop_periodic_task(t) for t in created_tasks))
+    await asyncio.gather(
+        *(stop_periodic_task(t, timeout=stop_task_timeout) for t in created_tasks)
+    )
 
 
 async def test_background_task_created_and_deleted(

packages/simcore-sdk/src/simcore_sdk/node_ports_common/filemanager.py

@@ -1,4 +1,5 @@
 import logging
+from asyncio import CancelledError
 from pathlib import Path
 from typing import Optional, Union
 
@@ -262,13 +263,15 @@ async def download_file_from_link(
 async def _abort_upload(
     session: ClientSession, upload_links: FileUploadSchema, *, reraise_exceptions: bool
 ) -> None:
+    # abort the upload correctly, so it can revert back to last version
     try:
         async with session.post(upload_links.links.abort_upload) as resp:
             resp.raise_for_status()
     except ClientError:
         log.warning("Error while aborting upload", exc_info=True)
         if reraise_exceptions:
             raise
+    log.warning("Upload aborted")
 
 
 async def upload_file(
@@ -352,10 +355,12 @@ async def upload_file(
         except (r_clone.RCloneFailedError, exceptions.S3TransferError) as exc:
             log.error("The upload failed with an unexpected error:", exc_info=True)
             if upload_links:
-                # abort the upload correctly, so it can revert back to last version
                 await _abort_upload(session, upload_links, reraise_exceptions=False)
-                log.warning("Upload aborted")
             raise exceptions.S3TransferError from exc
+        except CancelledError:
+            if upload_links:
+                await _abort_upload(session, upload_links, reraise_exceptions=False)
+            raise
         if io_log_redirect_cb:
             await io_log_redirect_cb(f"upload of {file_to_upload} complete.")
         return store_id, e_tag

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

@@ -91,18 +91,18 @@ async def containers_docker_status(
             return {}
 
     @log_decorator(logger=logger)
-    async def service_disable_dir_watcher(
+    async def disable_service_outputs_watcher(
         self, dynamic_sidecar_endpoint: AnyHttpUrl
     ) -> None:
-        await self._thin_client.patch_containers_directory_watcher(
+        await self._thin_client.patch_containers_outputs_watcher(
             dynamic_sidecar_endpoint, is_enabled=False
         )
 
     @log_decorator(logger=logger)
-    async def service_enable_dir_watcher(
+    async def enable_service_outputs_watcher(
         self, dynamic_sidecar_endpoint: AnyHttpUrl
     ) -> None:
-        await self._thin_client.patch_containers_directory_watcher(
+        await self._thin_client.patch_containers_outputs_watcher(
             dynamic_sidecar_endpoint, is_enabled=True
         )
 
@@ -367,11 +367,10 @@ async def pull_service_output_ports(
     async def push_service_output_ports(
         self,
         dynamic_sidecar_endpoint: AnyHttpUrl,
-        port_keys: Optional[list[str]] = None,
         progress_callback: Optional[ProgressCallback] = None,
     ) -> None:
         response = await self._thin_client.post_containers_tasks_ports_outputs_push(
-            dynamic_sidecar_endpoint, port_keys
+            dynamic_sidecar_endpoint
         )
         task_id: TaskId = response.json()
 

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

@@ -90,7 +90,7 @@ async def get_containers(
 
     @retry_on_errors
     @expect_status(status.HTTP_204_NO_CONTENT)
-    async def patch_containers_directory_watcher(
+    async def patch_containers_outputs_watcher(
         self, dynamic_sidecar_endpoint: AnyHttpUrl, *, is_enabled: bool
     ) -> Response:
         url = self._get_url(dynamic_sidecar_endpoint, "/containers/directory-watcher")
@@ -210,13 +210,10 @@ async def post_containers_tasks_ports_outputs_pull(
     @retry_on_errors
     @expect_status(status.HTTP_202_ACCEPTED)
     async def post_containers_tasks_ports_outputs_push(
-        self,
-        dynamic_sidecar_endpoint: AnyHttpUrl,
-        port_keys: Optional[list[str]] = None,
+        self, dynamic_sidecar_endpoint: AnyHttpUrl
     ) -> Response:
-        port_keys = [] if port_keys is None else port_keys
         url = self._get_url(dynamic_sidecar_endpoint, "/containers/ports/outputs:push")
-        return await self.client.post(url, json=port_keys)
+        return await self.client.post(url)
 
     @retry_on_errors
     @expect_status(status.HTTP_202_ACCEPTED)

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

@@ -1,7 +1,6 @@
 import logging
 from collections import deque
-from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Deque, Dict, Final, List, Optional, Type
+from typing import Any, Deque, Final, Optional
 
 from fastapi import FastAPI
 from pydantic import AnyHttpUrl
@@ -47,30 +46,7 @@
 RESOURCE_STATE_AND_INPUTS: Final[ResourceName] = "state_and_inputs"
 
 
-@asynccontextmanager
-async def disabled_directory_watcher(
-    dynamic_sidecar_client: DynamicSidecarClient, dynamic_sidecar_endpoint: AnyHttpUrl
-) -> AsyncIterator[None]:
-    """
-    The following will happen when using this context manager:
-    - Disables file system event watcher while writing
-        to the outputs directory to avoid data being pushed
-        via nodeports upon change.
-    - Enables file system event watcher so data from outputs
-        can be again synced via nodeports upon change.
-    """
-    try:
-        await dynamic_sidecar_client.service_disable_dir_watcher(
-            dynamic_sidecar_endpoint
-        )
-        yield
-    finally:
-        await dynamic_sidecar_client.service_enable_dir_watcher(
-            dynamic_sidecar_endpoint
-        )
-
-
-def get_repository(app: FastAPI, repo_type: Type[BaseRepository]) -> BaseRepository:
+def get_repository(app: FastAPI, repo_type: type[BaseRepository]) -> BaseRepository:
     return get_base_repository(engine=app.state.engine, repo_type=repo_type)
 
 
@@ -80,8 +56,8 @@ def get_director_v0_client(app: FastAPI) -> DirectorV0Client:
 
 
 def parse_containers_inspect(
-    containers_inspect: Optional[Dict[str, Any]]
-) -> List[DockerContainerInspect]:
+    containers_inspect: Optional[dict[str, Any]]
+) -> list[DockerContainerInspect]:
     results: Deque[DockerContainerInspect] = deque()
 
     if containers_inspect is None:
@@ -94,10 +70,10 @@ def parse_containers_inspect(
 
 
 def are_all_user_services_containers_running(
-    containers_inspect: List[DockerContainerInspect],
+    containers_inspect: list[DockerContainerInspect],
 ) -> bool:
     return len(containers_inspect) > 0 and all(
-        (x.status == DockerStatus.RUNNING for x in containers_inspect)
+        x.status == DockerStatus.RUNNING for x in containers_inspect
     )
 
 

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

@@ -59,7 +59,6 @@
     RESOURCE_STATE_AND_INPUTS,
     are_all_user_services_containers_running,
     attempt_pod_removal_and_data_saving,
-    disabled_directory_watcher,
     get_director_v0_client,
     get_repository,
     parse_containers_inspect,
@@ -362,46 +361,43 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         )
 
         async def _pull_outputs_and_state():
-            async with disabled_directory_watcher(
-                dynamic_sidecar_client, dynamic_sidecar_endpoint
-            ):
-                tasks = [
-                    dynamic_sidecar_client.pull_service_output_ports(
+            tasks = [
+                dynamic_sidecar_client.pull_service_output_ports(
+                    dynamic_sidecar_endpoint
+                )
+            ]
+            # When enabled no longer downloads state via nodeports
+            # S3 is used to store state paths
+            if not app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
+                tasks.append(
+                    dynamic_sidecar_client.restore_service_state(
                         dynamic_sidecar_endpoint
                     )
-                ]
-                # When enabled no longer downloads state via nodeports
-                # S3 is used to store state paths
-                if not app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
-                    tasks.append(
-                        dynamic_sidecar_client.restore_service_state(
-                            dynamic_sidecar_endpoint
-                        )
-                    )
+                )
 
-                await logged_gather(*tasks, max_concurrency=2)
+            await logged_gather(*tasks, max_concurrency=2)
 
-                # inside this directory create the missing dirs, fetch those form the labels
-                director_v0_client: DirectorV0Client = get_director_v0_client(app)
-                simcore_service_labels: SimcoreServiceLabels = (
-                    await director_v0_client.get_service_labels(
-                        service=ServiceKeyVersion(
-                            key=scheduler_data.key, version=scheduler_data.version
-                        )
+            # inside this directory create the missing dirs, fetch those form the labels
+            director_v0_client: DirectorV0Client = get_director_v0_client(app)
+            simcore_service_labels: SimcoreServiceLabels = (
+                await director_v0_client.get_service_labels(
+                    service=ServiceKeyVersion(
+                        key=scheduler_data.key, version=scheduler_data.version
                     )
                 )
-                service_outputs_labels = json.loads(
-                    simcore_service_labels.dict().get("io.simcore.outputs", "{}")
-                ).get("outputs", {})
-                logger.debug(
-                    "Creating dirs from service outputs labels: %s",
-                    service_outputs_labels,
-                )
-                await dynamic_sidecar_client.service_outputs_create_dirs(
-                    dynamic_sidecar_endpoint, service_outputs_labels
-                )
+            )
+            service_outputs_labels = json.loads(
+                simcore_service_labels.dict().get("io.simcore.outputs", "{}")
+            ).get("outputs", {})
+            logger.debug(
+                "Creating dirs from service outputs labels: %s",
+                service_outputs_labels,
+            )
+            await dynamic_sidecar_client.service_outputs_create_dirs(
+                dynamic_sidecar_endpoint, service_outputs_labels
+            )
 
-                scheduler_data.dynamic_sidecar.is_service_environment_ready = True
+            scheduler_data.dynamic_sidecar.is_service_environment_ready = True
 
         if dynamic_sidecar_settings.DYNAMIC_SIDECAR_DOCKER_NODE_RESOURCE_LIMITS_ENABLED:
             node_rights_manager = NodeRightsManager.instance(app)
@@ -481,6 +477,10 @@ async def progress_create_containers(
             dynamic_sidecar_endpoint, compose_spec, progress_create_containers
         )
 
+        await dynamic_sidecar_client.enable_service_outputs_watcher(
+            dynamic_sidecar_endpoint
+        )
+
         # Starts PROXY -----------------------------------------------
         # The entrypoint container name was now computed
         # continue starting the proxy