send logs to parent node as well

ITISFoundation · Nov 21, 2023 · 812e26f · 812e26f
1 parent 61a672f
commit 812e26f
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 8 deletions.
diff --git a/packages/dask-task-models-library/src/dask_task_models_library/container_tasks/protocol.py b/packages/dask-task-models-library/src/dask_task_models_library/container_tasks/protocol.py
@@ -28,6 +28,10 @@ class TaskOwner(BaseModel):
     parent_project_id: ProjectID | None
     parent_node_id: NodeID | None
 
+    @property
+    def has_parent(self) -> bool:
+        return bool(self.parent_node_id and self.parent_project_id)
+
     @root_validator
     @classmethod
     def check_parent_valid(cls, values: dict[str, Any]) -> dict[str, Any]:

diff --git a/...ices/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/dask_scheduler.py b/...ices/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/dask_scheduler.py
@@ -323,10 +323,9 @@ async def _task_progress_change_handler(self, event: str) -> None:
         with log_catch(_logger, reraise=False):
             task_progress_event = TaskProgressEvent.parse_raw(event)
             _logger.debug("received task progress update: %s", task_progress_event)
-            *_, user_id, project_id, node_id = parse_dask_job_id(
-                task_progress_event.job_id
-            )
-
+            user_id = task_progress_event.task_owner.user_id
+            project_id = task_progress_event.task_owner.project_id
+            node_id = task_progress_event.task_owner.node_id
             comp_tasks_repo = CompTasksRepository(self.db_engine)
             task = await comp_tasks_repo.get_task(project_id, node_id)
             if task.progress is None:
@@ -355,12 +354,22 @@ async def _task_log_change_handler(self, event: str) -> None:
         with log_catch(_logger, reraise=False):
             task_log_event = TaskLogEvent.parse_raw(event)
             _logger.debug("received task log update: %s", task_log_event)
-            *_, user_id, project_id, node_id = parse_dask_job_id(task_log_event.job_id)
             await publish_service_log(
                 self.rabbitmq_client,
-                user_id=user_id,
-                project_id=project_id,
-                node_id=node_id,
+                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,
+                )