removed wake_up callback from api interface

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_base.py

@@ -76,6 +76,22 @@
 _MAX_WAITING_FOR_CLUSTER_TIMEOUT_IN_MIN: Final[int] = 10
 
 
+def _temporary_empty_wake_up_callack(
+    user_id: UserID, project_id: ProjectID, iteration: Iteration
+) -> Callable[[], None]:
+    def _cb() -> None:
+        ...
+
+    # async def _async_cb():
+    #     db_engine = get_db_engine(app)
+    #     rabbit_mq_client = get_rabbitmq_client(app)
+    #     comp_run = await CompRunsRepository.instance(db_engine).get(
+    #         user_id=user_id, project_id=project_id, iteration=iteration
+    #     )
+    #     await request_pipeline_scheduling(comp_run, rabbit_mq_client, db_engine)
+    return _cb
+
+
 @dataclass(frozen=True, slots=True)
 class SortedTasks:
     started: list[CompTaskAtDB]
@@ -496,9 +512,6 @@ async def schedule_pipeline(
         user_id: UserID,
         project_id: ProjectID,
         iteration: Iteration,
-        wake_up_callback: Callable[
-            [], None
-        ],  # TODO: this should not be in the interface
     ) -> None:
         """schedules a pipeline for a given user, project and iteration.
 
@@ -537,7 +550,9 @@ async def schedule_pipeline(
                         comp_tasks=comp_tasks,
                         dag=dag,
                         comp_run=comp_run,
-                        wake_up_callback=wake_up_callback,
+                        wake_up_callback=_temporary_empty_wake_up_callack(
+                            user_id, project_id, iteration
+                        ),
                     )
                 # 4. timeout if waiting for cluster has been there for more than X minutes
                 comp_tasks = await self._timeout_if_waiting_for_cluster_too_long(

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_worker.py

@@ -1,36 +1,18 @@
 import functools
 import logging
-from typing import Callable, cast
+from typing import cast
 
 from fastapi import FastAPI
-from models_library.projects import ProjectID
-from models_library.users import UserID
 from servicelib.logging_utils import log_context
 
 from ..rabbitmq import get_rabbitmq_client
-from ._models import Iteration, SchedulePipelineRabbitMessage
+from ._models import SchedulePipelineRabbitMessage
 from ._scheduler_base import BaseCompScheduler
 from ._scheduler_factory import create_scheduler
 
 _logger = logging.getLogger(__name__)
 
 
-def _empty_wake_up_callack(
-    app: FastAPI, user_id: UserID, project_id: ProjectID, iteration: Iteration
-) -> Callable[[], None]:
-    def _cb() -> None:
-        ...
-
-    # async def _async_cb():
-    #     db_engine = get_db_engine(app)
-    #     rabbit_mq_client = get_rabbitmq_client(app)
-    #     comp_run = await CompRunsRepository.instance(db_engine).get(
-    #         user_id=user_id, project_id=project_id, iteration=iteration
-    #     )
-    #     await request_pipeline_scheduling(comp_run, rabbit_mq_client, db_engine)
-    return _cb
-
-
 def _get_scheduler_worker(app: FastAPI) -> BaseCompScheduler:
     return cast(BaseCompScheduler, app.state.scheduler_worker)
 
@@ -43,12 +25,6 @@ async def _handle_distributed_pipeline(app: FastAPI, data: bytes) -> bool:
             user_id=to_schedule_pipeline.user_id,
             project_id=to_schedule_pipeline.project_id,
             iteration=to_schedule_pipeline.iteration,
-            wake_up_callback=_empty_wake_up_callack(
-                app,
-                to_schedule_pipeline.user_id,
-                to_schedule_pipeline.project_id,
-                to_schedule_pipeline.iteration,
-            ),
         )
         return True
 

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_scheduler_dask.py

@@ -173,7 +173,6 @@ async def test_broken_pipeline_configuration_is_not_scheduled_and_aborted(
     fake_workbench_adjacency: dict[str, Any],
     sqlalchemy_async_engine: AsyncEngine,
     run_metadata: RunMetadataDict,
-    mocked_wake_up_callback: mock.Mock,
 ):
     """A pipeline which comp_tasks are missing should not be scheduled.
     It shall be aborted and shown as such in the comp_runs db"""
@@ -211,10 +210,7 @@ async def test_broken_pipeline_configuration_is_not_scheduled_and_aborted(
         user_id=run_entry.user_id,
         project_id=run_entry.project_uuid,
         iteration=run_entry.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
-    # the pipeline is misconfigured, so the callback will NOT be called since nothing ran
-    mocked_wake_up_callback.assert_not_called()
 
     # check the database entry is correctly updated
     await assert_comp_runs(
@@ -270,7 +266,6 @@ async def _assert_schedule_pipeline_PENDING(  # noqa: N802
     published_tasks: list[CompTaskAtDB],
     mocked_dask_client: mock.MagicMock,
     scheduler: BaseCompScheduler,
-    wake_up_callback: Callable[[], None],
 ) -> list[CompTaskAtDB]:
     expected_pending_tasks = [
         published_tasks[1],
@@ -288,7 +283,6 @@ async def _return_tasks_pending(job_ids: list[str]) -> list[DaskClientTaskState]
         user_id=published_project.project.prj_owner,
         project_id=published_project.project.uuid,
         iteration=1,
-        wake_up_callback=wake_up_callback,
     )
     _assert_dask_client_correctly_initialized(mocked_dask_client, scheduler)
     await assert_comp_runs(
@@ -340,7 +334,6 @@ async def _return_tasks_pending(job_ids: list[str]) -> list[DaskClientTaskState]
         user_id=published_project.project.prj_owner,
         project_id=published_project.project.uuid,
         iteration=1,
-        wake_up_callback=wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -485,7 +478,6 @@ async def test_proper_pipeline_is_scheduled(  # noqa: PLR0915
     instrumentation_rabbit_client_parser: mock.AsyncMock,
     resource_tracking_rabbit_client_parser: mock.AsyncMock,
     run_metadata: RunMetadataDict,
-    mocked_wake_up_callback: mock.Mock,
 ):
     _mock_send_computation_tasks(published_project.tasks, mocked_dask_client)
 
@@ -504,7 +496,6 @@ async def test_proper_pipeline_is_scheduled(  # noqa: PLR0915
         expected_published_tasks,
         mocked_dask_client,
         scheduler,
-        mocked_wake_up_callback,
     )
 
     # -------------------------------------------------------------------------------
@@ -528,7 +519,6 @@ async def _return_1st_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -586,7 +576,6 @@ async def _return_1st_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     # comp_run, the comp_task switch to STARTED
     await assert_comp_runs(
@@ -663,7 +652,6 @@ async def _return_random_task_result(job_id) -> TaskOutputData:
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -771,7 +759,6 @@ async def _return_2nd_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -826,7 +813,6 @@ async def _return_2nd_task_failed(job_ids: list[str]) -> list[DaskClientTaskStat
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -886,7 +872,6 @@ async def _return_3rd_task_success(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await assert_comp_runs(
         sqlalchemy_async_engine,
@@ -938,7 +923,6 @@ async def test_task_progress_triggers(
     mocked_parse_output_data_fct: None,
     mocked_clean_task_output_and_log_files_if_invalid: None,
     run_metadata: RunMetadataDict,
-    mocked_wake_up_callback: mock.Mock,
 ):
     _mock_send_computation_tasks(published_project.tasks, mocked_dask_client)
     _run_in_db, expected_published_tasks = await _assert_start_pipeline(
@@ -955,7 +939,6 @@ async def test_task_progress_triggers(
         expected_published_tasks,
         mocked_dask_client,
         scheduler,
-        mocked_wake_up_callback,
     )
 
     # send some progress
@@ -1008,7 +991,6 @@ async def test_handling_of_disconnected_scheduler_dask(
     published_project: PublishedProject,
     backend_error: SchedulerError,
     run_metadata: RunMetadataDict,
-    mocked_wake_up_callback: mock.Mock,
 ):
     # this will create a non connected backend issue that will trigger re-connection
     mocked_dask_client_send_task = mocker.patch(
@@ -1060,7 +1042,6 @@ async def test_handling_of_disconnected_scheduler_dask(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     # after this step the tasks are marked as ABORTED
     await assert_comp_tasks(
@@ -1079,7 +1060,6 @@ async def test_handling_of_disconnected_scheduler_dask(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     # now the run should be ABORTED
     await assert_comp_runs(
@@ -1181,7 +1161,6 @@ async def test_handling_scheduling_after_reboot(
     mocked_parse_output_data_fct: mock.MagicMock,
     mocked_clean_task_output_fct: mock.MagicMock,
     reboot_state: RebootState,
-    mocked_wake_up_callback: mock.Mock,
 ):
     """After the dask client is rebooted, or that the director-v2 reboots the dv-2 internal scheduler
     shall continue scheduling correctly. Even though the task might have continued to run
@@ -1203,7 +1182,6 @@ async def mocked_get_task_result(_job_id: str) -> TaskOutputData:
         user_id=running_project.project.prj_owner,
         project_id=running_project.project.uuid,
         iteration=1,
-        wake_up_callback=mocked_wake_up_callback,
     )
     # the status will be called once for all RUNNING tasks
     mocked_dask_client.get_tasks_status.assert_called_once()
@@ -1275,7 +1253,6 @@ async def test_handling_cancellation_of_jobs_after_reboot(
     scheduler: BaseCompScheduler,
     mocked_parse_output_data_fct: mock.MagicMock,
     mocked_clean_task_output_fct: mock.MagicMock,
-    mocked_wake_up_callback: mock.Mock,
 ):
     """A running pipeline was cancelled by a user and the DV-2 was restarted BEFORE
     It could actually cancel the task. On reboot the DV-2 shall recover
@@ -1314,7 +1291,6 @@ async def mocked_get_tasks_status(job_ids: list[str]) -> list[DaskClientTaskStat
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     mocked_dask_client.abort_computation_task.assert_called()
     assert mocked_dask_client.abort_computation_task.call_count == len(
@@ -1364,7 +1340,6 @@ async def _return_random_task_result(job_id) -> TaskOutputData:
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     # now should be stopped
     await assert_comp_tasks(
@@ -1410,7 +1385,6 @@ async def test_running_pipeline_triggers_heartbeat(
     published_project: PublishedProject,
     resource_tracking_rabbit_client_parser: mock.AsyncMock,
     run_metadata: RunMetadataDict,
-    mocked_wake_up_callback: mock.Mock,
 ):
     _mock_send_computation_tasks(published_project.tasks, mocked_dask_client)
     run_in_db, expected_published_tasks = await _assert_start_pipeline(
@@ -1427,7 +1401,6 @@ async def test_running_pipeline_triggers_heartbeat(
         expected_published_tasks,
         mocked_dask_client,
         scheduler,
-        mocked_wake_up_callback,
     )
     # -------------------------------------------------------------------------------
     # 2. the "worker" starts processing a task
@@ -1458,7 +1431,6 @@ async def _return_1st_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
 
     messages = await _assert_message_received(
@@ -1475,13 +1447,11 @@ async def _return_1st_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await scheduler.schedule_pipeline(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     messages = await _assert_message_received(
         resource_tracking_rabbit_client_parser,
@@ -1497,13 +1467,11 @@ async def _return_1st_task_running(job_ids: list[str]) -> list[DaskClientTaskSta
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     await scheduler.schedule_pipeline(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     messages = await _assert_message_received(
         resource_tracking_rabbit_client_parser,
@@ -1531,7 +1499,6 @@ async def test_pipeline_with_on_demand_cluster_with_not_ready_backend_waits(
     run_metadata: RunMetadataDict,
     mocked_get_or_create_cluster: mock.Mock,
     faker: Faker,
-    mocked_wake_up_callback: mock.Mock,
 ):
     mocked_get_or_create_cluster.side_effect = (
         ComputationalBackendOnDemandNotReadyError(
@@ -1578,7 +1545,6 @@ async def test_pipeline_with_on_demand_cluster_with_not_ready_backend_waits(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     mocked_get_or_create_cluster.assert_called()
     assert mocked_get_or_create_cluster.call_count == 1
@@ -1604,7 +1570,6 @@ async def test_pipeline_with_on_demand_cluster_with_not_ready_backend_waits(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     mocked_get_or_create_cluster.assert_called()
     assert mocked_get_or_create_cluster.call_count == 1
@@ -1641,7 +1606,6 @@ async def test_pipeline_with_on_demand_cluster_with_no_clusters_keeper_fails(
     run_metadata: RunMetadataDict,
     mocked_get_or_create_cluster: mock.Mock,
     get_or_create_exception: Exception,
-    mocked_wake_up_callback: mock.Mock,
 ):
     mocked_get_or_create_cluster.side_effect = get_or_create_exception
     # running the pipeline will trigger a call to the clusters-keeper
@@ -1683,7 +1647,6 @@ async def test_pipeline_with_on_demand_cluster_with_no_clusters_keeper_fails(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     mocked_get_or_create_cluster.assert_called()
     assert mocked_get_or_create_cluster.call_count == 1
@@ -1709,7 +1672,6 @@ async def test_pipeline_with_on_demand_cluster_with_no_clusters_keeper_fails(
         user_id=run_in_db.user_id,
         project_id=run_in_db.project_uuid,
         iteration=run_in_db.iteration,
-        wake_up_callback=mocked_wake_up_callback,
     )
     mocked_get_or_create_cluster.assert_not_called()
     await assert_comp_runs(

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_worker.py

@@ -67,5 +67,4 @@ async def test_worker_properly_calls_scheduler_api(
         user_id=published_project.project.prj_owner,
         project_id=published_project.project.uuid,
         iteration=1,
-        wake_up_callback=mock.ANY,
     )