Merge branch 'master' into 2024/fix/tracing

Author: mrnicegyu11

.codecov.yml

@@ -18,6 +18,7 @@ flag_management:
 
 component_management:
   default_rules:
+    carryforward: true
     statuses:
       - type: project
         target: auto

.github/workflows/ci-testing-deploy.yml

@@ -450,6 +450,51 @@ jobs:
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
 
+  unit-test-webserver-04:
+      needs: changes
+      if: ${{ needs.changes.outputs.webserver == 'true' || github.event_name == 'push' }}
+      timeout-minutes: 25 # if this timeout gets too small, then split the tests
+      name: "[unit] webserver 04"
+      runs-on: ${{ matrix.os }}
+      strategy:
+        matrix:
+          python: ["3.11"]
+          os: [ubuntu-22.04]
+        fail-fast: false
+      steps:
+        - uses: actions/checkout@v4
+        - name: setup docker buildx
+          id: buildx
+          uses: docker/setup-buildx-action@v3
+          with:
+            driver: docker-container
+        - name: setup python environment
+          uses: actions/setup-python@v5
+          with:
+            python-version: ${{ matrix.python }}
+        - name: install uv
+          uses: astral-sh/setup-uv@v3
+          with:
+            version: "0.4.x"
+            enable-cache: false
+            cache-dependency-glob: "**/web/server/requirements/ci.txt"
+        - name: show system version
+          run: ./ci/helpers/show_system_versions.bash
+        - name: install webserver
+          run: ./ci/github/unit-testing/webserver.bash install
+        - name: test
+          run: ./ci/github/unit-testing/webserver.bash test_with_db 04
+        - uses: codecov/[email protected]
+          env:
+            CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+          with:
+            flags: unittests #optional
+        - name: Upload test results to Codecov
+          if: ${{ !cancelled() }}
+          uses: codecov/test-results-action@v1
+          with:
+            token: ${{ secrets.CODECOV_TOKEN }}
+
   unit-test-storage:
     needs: changes
     if: ${{ needs.changes.outputs.storage == 'true' || github.event_name == 'push' }}
@@ -1875,6 +1920,7 @@ jobs:
         unit-test-webserver-01,
         unit-test-webserver-02,
         unit-test-webserver-03,
+        unit-test-webserver-04,
       ]
     runs-on: ubuntu-latest
     steps:

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

@@ -25,15 +25,36 @@ class PeriodicTaskCancellationError(PydanticErrorMixin, Exception):
     msg_template: str = "Could not cancel task '{task_name}'"
 
 
+class SleepUsingAsyncioEvent:
+    """Sleep strategy that waits on an event to be set."""
+
+    def __init__(self, event: "asyncio.Event") -> None:
+        self.event = event
+
+    async def __call__(self, timeout: float | None) -> None:
+        with contextlib.suppress(TimeoutError):
+            await asyncio.wait_for(self.event.wait(), timeout=timeout)
+            self.event.clear()
+
+
 async def _periodic_scheduled_task(
     task: Callable[..., Awaitable[None]],
     *,
     interval: datetime.timedelta,
     task_name: str,
+    early_wake_up_event: asyncio.Event | None,
     **task_kwargs,
 ) -> None:
     # NOTE: This retries forever unless cancelled
-    async for attempt in AsyncRetrying(wait=wait_fixed(interval.total_seconds())):
+    nap = (
+        asyncio.sleep
+        if early_wake_up_event is None
+        else SleepUsingAsyncioEvent(early_wake_up_event)
+    )
+    async for attempt in AsyncRetrying(
+        sleep=nap,
+        wait=wait_fixed(interval.total_seconds()),
+    ):
         with attempt:
             with log_context(
                 _logger,
@@ -51,6 +72,7 @@ def start_periodic_task(
     interval: datetime.timedelta,
     task_name: str,
     wait_before_running: datetime.timedelta = datetime.timedelta(0),
+    early_wake_up_event: asyncio.Event | None = None,
     **kwargs,
 ) -> asyncio.Task:
     with log_context(
@@ -64,6 +86,7 @@ def start_periodic_task(
                 task,
                 interval=interval,
                 task_name=task_name,
+                early_wake_up_event=early_wake_up_event,
                 **kwargs,
             ),
             name=task_name,

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/resource_usage_tracker/errors.py

@@ -6,7 +6,8 @@
 
 import asyncio
 import datetime
-from typing import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
+from typing import Final
 from unittest import mock
 
 import pytest
@@ -18,38 +19,50 @@
     stop_periodic_task,
 )
 
-_FAST_POLL_INTERVAL = 1
+_FAST_POLL_INTERVAL: Final[int] = 1
+_VERY_SLOW_POLL_INTERVAL: Final[int] = 100
 
 
 @pytest.fixture
 def mock_background_task(mocker: MockerFixture) -> mock.AsyncMock:
-    mocked_task = mocker.AsyncMock(return_value=None)
-    return mocked_task
+    return mocker.AsyncMock(return_value=None)
 
 
 @pytest.fixture
 def task_interval() -> datetime.timedelta:
     return datetime.timedelta(seconds=_FAST_POLL_INTERVAL)
 
 
-@pytest.fixture(params=[None, 1])
+@pytest.fixture
+def very_long_task_interval() -> datetime.timedelta:
+    return datetime.timedelta(seconds=_VERY_SLOW_POLL_INTERVAL)
+
+
+@pytest.fixture(params=[None, 1], ids=lambda x: f"stop-timeout={x}")
 def stop_task_timeout(request: pytest.FixtureRequest) -> float | None:
     return request.param
 
 
 @pytest.fixture
 async def create_background_task(
     faker: Faker, stop_task_timeout: float | None
-) -> AsyncIterator[Callable[[datetime.timedelta, Callable], Awaitable[asyncio.Task]]]:
+) -> AsyncIterator[
+    Callable[
+        [datetime.timedelta, Callable, asyncio.Event | None], Awaitable[asyncio.Task]
+    ]
+]:
     created_tasks = []
 
     async def _creator(
-        interval: datetime.timedelta, task: Callable[..., Awaitable]
+        interval: datetime.timedelta,
+        task: Callable[..., Awaitable],
+        early_wake_up_event: asyncio.Event | None,
     ) -> asyncio.Task:
         background_task = start_periodic_task(
             task,
             interval=interval,
             task_name=faker.pystr(),
+            early_wake_up_event=early_wake_up_event,
         )
         assert background_task
         created_tasks.append(background_task)
@@ -62,33 +75,69 @@ async def _creator(
     )
 
 
+@pytest.mark.parametrize(
+    "wake_up_event", [None, asyncio.Event], ids=lambda x: f"wake-up-event: {x}"
+)
 async def test_background_task_created_and_deleted(
     mock_background_task: mock.AsyncMock,
     task_interval: datetime.timedelta,
     create_background_task: Callable[
-        [datetime.timedelta, Callable], Awaitable[asyncio.Task]
+        [datetime.timedelta, Callable, asyncio.Event | None], Awaitable[asyncio.Task]
     ],
+    wake_up_event: Callable | None,
 ):
-    task = await create_background_task(
+    event = wake_up_event() if wake_up_event else None
+    _task = await create_background_task(
         task_interval,
         mock_background_task,
+        event,
     )
     await asyncio.sleep(5 * task_interval.total_seconds())
     mock_background_task.assert_called()
-    assert mock_background_task.call_count > 1
+    assert mock_background_task.call_count > 2
+
+
+async def test_background_task_wakes_up_early(
+    mock_background_task: mock.AsyncMock,
+    very_long_task_interval: datetime.timedelta,
+    create_background_task: Callable[
+        [datetime.timedelta, Callable, asyncio.Event | None], Awaitable[asyncio.Task]
+    ],
+):
+    wake_up_event = asyncio.Event()
+    _task = await create_background_task(
+        very_long_task_interval,
+        mock_background_task,
+        wake_up_event,
+    )
+    await asyncio.sleep(5 * _FAST_POLL_INTERVAL)
+    # now the task should have run only once
+    mock_background_task.assert_called_once()
+    await asyncio.sleep(5 * _FAST_POLL_INTERVAL)
+    mock_background_task.assert_called_once()
+    # this should wake up the task
+    wake_up_event.set()
+    await asyncio.sleep(5 * _FAST_POLL_INTERVAL)
+    mock_background_task.assert_called()
+    assert mock_background_task.call_count == 2
+    # no change this now waits again a very long time
+    await asyncio.sleep(5 * _FAST_POLL_INTERVAL)
+    mock_background_task.assert_called()
+    assert mock_background_task.call_count == 2
 
 
 async def test_background_task_raises_restarts(
     mock_background_task: mock.AsyncMock,
     task_interval: datetime.timedelta,
     create_background_task: Callable[
-        [datetime.timedelta, Callable], Awaitable[asyncio.Task]
+        [datetime.timedelta, Callable, asyncio.Event | None], Awaitable[asyncio.Task]
     ],
 ):
     mock_background_task.side_effect = RuntimeError("pytest faked runtime error")
-    task = await create_background_task(
+    _task = await create_background_task(
         task_interval,
         mock_background_task,
+        None,
     )
     await asyncio.sleep(5 * task_interval.total_seconds())
     mock_background_task.assert_called()
@@ -99,13 +148,14 @@ async def test_background_task_correctly_cancels(
     mock_background_task: mock.AsyncMock,
     task_interval: datetime.timedelta,
     create_background_task: Callable[
-        [datetime.timedelta, Callable], Awaitable[asyncio.Task]
+        [datetime.timedelta, Callable, asyncio.Event | None], Awaitable[asyncio.Task]
     ],
 ):
     mock_background_task.side_effect = asyncio.CancelledError
-    task = await create_background_task(
+    _task = await create_background_task(
         task_interval,
         mock_background_task,
+        None,
     )
     await asyncio.sleep(5 * task_interval.total_seconds())
     # the task will be called once, and then stop

packages/service-library/tests/test_background_task.py

@@ -48,7 +48,7 @@ def _get_project(connection: Connection, project_uuid: UUID) -> ResultProxy:
 def _get_hidden_project(connection: Connection, prj_owner: int) -> ResultProxy:
     return connection.execute(
         select(projects).where(
-            and_(projects.c.prj_owner == prj_owner, projects.c.hidden == True)
+            and_(projects.c.prj_owner == prj_owner, projects.c.hidden.is_(True))
         )
     )
 
@@ -61,7 +61,7 @@ def _get_file_meta_data_without_soft_links(
             and_(
                 file_meta_data.c.node_id == f"{node_uuid}",
                 file_meta_data.c.project_id == f"{project_id}",
-                file_meta_data.c.is_soft_link != True,
+                file_meta_data.c.is_soft_link.is_not(True),
             )
         )
     )

scripts/maintenance/migrate_project/src/db.py

@@ -1,7 +1,7 @@
 from fastapi import Depends, FastAPI, Request
 
 from ...core.settings import ComputationalBackendSettings
-from ...modules.comp_scheduler.base_scheduler import BaseCompScheduler
+from ...modules.comp_scheduler import BaseCompScheduler
 from . import get_app
 
 

services/director-v2/src/simcore_service_director_v2/api/dependencies/scheduler.py

@@ -63,7 +63,7 @@
 from ...models.comp_runs import CompRunsAtDB, ProjectMetadataDict, RunMetadataDict
 from ...models.comp_tasks import CompTaskAtDB
 from ...modules.catalog import CatalogClient
-from ...modules.comp_scheduler.base_scheduler import BaseCompScheduler
+from ...modules.comp_scheduler import BaseCompScheduler
 from ...modules.db.repositories.clusters import ClustersRepository
 from ...modules.db.repositories.comp_pipelines import CompPipelinesRepository
 from ...modules.db.repositories.comp_runs import CompRunsRepository
@@ -288,7 +288,7 @@ async def _try_start_pipeline(
 )
 # NOTE: in case of a burst of calls to that endpoint, we might end up in a weird state.
 @run_sequentially_in_context(target_args=["computation.project_id"])
-async def create_computation(  # noqa: PLR0913  # pylint:disable=too-many-positional-arguments
+async def create_computation(  # noqa: PLR0913 # pylint: disable=too-many-positional-arguments
     computation: ComputationCreate,
     request: Request,
     project_repo: Annotated[

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

@@ -173,9 +173,9 @@ def init_app(settings: AppSettings | None = None) -> FastAPI:
     )
     if dynamic_scheduler_enabled or computational_backend_enabled:
         rabbitmq.setup(app)
+        redis.setup(app)
 
     if dynamic_scheduler_enabled:
-        redis.setup(app)
         dynamic_sidecar.setup(app)
         socketio.setup(app)
         notifier.setup(app)

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

@@ -1,3 +1,15 @@
-from .background_task import setup
+from fastapi import FastAPI
 
-__all__: tuple[str, ...] = ("setup",)
+from ._base_scheduler import BaseCompScheduler
+from ._task import on_app_shutdown, on_app_startup
+
+
+def setup(app: FastAPI):
+    app.add_event_handler("startup", on_app_startup(app))
+    app.add_event_handler("shutdown", on_app_shutdown(app))
+
+
+__all__: tuple[str, ...] = (
+    "setup",
+    "BaseCompScheduler",
+)