removed logs sub

Author: sanderegg

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

@@ -8,7 +8,6 @@
 import arrow
 from dask_task_models_library.container_tasks.errors import TaskCancelledError
 from dask_task_models_library.container_tasks.events import (
-    TaskLogEvent,
     TaskProgressEvent,
 )
 from dask_task_models_library.container_tasks.io import TaskOutputData
@@ -38,7 +37,6 @@
 )
 from ...utils.dask_client_utils import TaskHandlers
 from ...utils.rabbitmq import (
-    publish_service_log,
     publish_service_progress,
     publish_service_resource_tracking_stopped,
     publish_service_stopped_metrics,
@@ -92,7 +90,6 @@ def __post_init__(self) -> None:
         self.dask_clients_pool.register_handlers(
             TaskHandlers(
                 self._task_progress_change_handler,
-                self._task_log_change_handler,
             )
         )
 
@@ -378,27 +375,3 @@ async def _task_progress_change_handler(self, event: str) -> None:
                 node_id=node_id,
                 progress=task_progress_event.progress,
             )
-
-    async def _task_log_change_handler(self, event: str) -> None:
-        with log_catch(_logger, reraise=False):
-            task_log_event = TaskLogEvent.model_validate_json(event)
-            _logger.debug("received task log update: %s", task_log_event)
-            await publish_service_log(
-                self.rabbitmq_client,
-                user_id=task_log_event.task_owner.user_id,
-                project_id=task_log_event.task_owner.project_id,
-                node_id=task_log_event.task_owner.node_id,
-                log=task_log_event.log,
-                log_level=task_log_event.log_level,
-            )
-            if task_log_event.task_owner.has_parent:
-                assert task_log_event.task_owner.parent_project_id  # nosec
-                assert task_log_event.task_owner.parent_node_id  # nosec
-                await publish_service_log(
-                    self.rabbitmq_client,
-                    user_id=task_log_event.task_owner.user_id,
-                    project_id=task_log_event.task_owner.parent_project_id,
-                    node_id=task_log_event.task_owner.parent_node_id,
-                    log=task_log_event.log,
-                    log_level=task_log_event.log_level,
-                )

services/director-v2/src/simcore_service_director_v2/modules/dask_client.py

@@ -187,7 +187,6 @@ async def delete(self) -> None:
     def register_handlers(self, task_handlers: TaskHandlers) -> None:
         _event_consumer_map = [
             (self.backend.progress_sub, task_handlers.task_progress_handler),
-            (self.backend.logs_sub, task_handlers.task_log_handler),
         ]
         self._subscribed_tasks = [
             asyncio.create_task(

services/director-v2/src/simcore_service_director_v2/utils/dask_client_utils.py

@@ -6,7 +6,6 @@
 
 import distributed
 from dask_task_models_library.container_tasks.events import (
-    TaskLogEvent,
     TaskProgressEvent,
 )
 from models_library.clusters import ClusterAuthentication, TLSAuthentication
@@ -19,7 +18,6 @@
 @dataclass
 class TaskHandlers:
     task_progress_handler: Callable[[str], Awaitable[None]]
-    task_log_handler: Callable[[str], Awaitable[None]]
 
 
 logger = logging.getLogger(__name__)
@@ -30,13 +28,11 @@ class DaskSubSystem:
     client: distributed.Client
     scheduler_id: str
     progress_sub: distributed.Sub = field(init=False)
-    logs_sub: distributed.Sub = field(init=False)
 
     def __post_init__(self) -> None:
         self.progress_sub = distributed.Sub(
             TaskProgressEvent.topic_name(), client=self.client
         )
-        self.logs_sub = distributed.Sub(TaskLogEvent.topic_name(), client=self.client)
 
     async def close(self) -> None:
         # NOTE: if the Sub are deleted before closing the connection,

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

@@ -1078,9 +1078,7 @@ def fake_remote_fct(
 
 @pytest.fixture
 async def fake_task_handlers(mocker: MockerFixture) -> TaskHandlers:
-    return TaskHandlers(
-        task_progress_handler=mocker.MagicMock(), task_log_handler=mocker.MagicMock()
-    )
+    return TaskHandlers(task_progress_handler=mocker.MagicMock())
 
 
 async def test_dask_sub_handlers(
@@ -1154,7 +1152,6 @@ def fake_remote_fct(
             fake_task_handlers.task_progress_handler.assert_called_with(
                 "my name is progress"
             )
-            fake_task_handlers.task_log_handler.assert_called_with("my name is logs")
     await _assert_wait_for_cb_call(mocked_user_completed_cb)