cleanup

Author: sanderegg

services/dask-sidecar/src/simcore_service_dask_sidecar/computational_sidecar/core.py

@@ -160,7 +160,7 @@ async def _retrieve_output_data(
     async def _publish_sidecar_log(
         self, log: LogMessageStr, log_level: LogLevelInt = logging.INFO
     ) -> None:
-        self.task_publishers.publish_logs(
+        await self.task_publishers.publish_logs(
             message=f"[sidecar] {log}", log_level=log_level
         )
 
@@ -174,14 +174,18 @@ async def run(self, command: list[str]) -> TaskOutputData:
 
         settings = ApplicationSettings.create_from_envs()
         run_id = f"{uuid4()}"
-        async with Docker() as docker_client, TaskSharedVolumes(
-            Path(f"{settings.SIDECAR_COMP_SERVICES_SHARED_FOLDER}/{run_id}")
-        ) as task_volumes, ProgressBarData(
-            num_steps=3,
-            step_weights=[5 / 100, 90 / 100, 5 / 100],
-            progress_report_cb=self.task_publishers.publish_progress,
-            description="running",
-        ) as progress_bar:
+        async with (
+            Docker() as docker_client,
+            TaskSharedVolumes(
+                Path(f"{settings.SIDECAR_COMP_SERVICES_SHARED_FOLDER}/{run_id}")
+            ) as task_volumes,
+            ProgressBarData(
+                num_steps=3,
+                step_weights=[5 / 100, 90 / 100, 5 / 100],
+                progress_report_cb=self.task_publishers.publish_progress,
+                description="running",
+            ) as progress_bar,
+        ):
             # PRE-PROCESSING
             await pull_image(
                 docker_client,
@@ -216,24 +220,28 @@ async def run(self, command: list[str]) -> TaskOutputData:
             )
             await progress_bar.update()  # NOTE:  (1 step weighting 5%)
             # PROCESSING (1 step weighted 90%)
-            async with managed_container(
-                docker_client,
-                config,
-                name=f"{self.task_parameters.image.split(sep='/')[-1]}_{run_id}",
-            ) as container, progress_bar.sub_progress(
-                100, description="processing"
-            ) as processing_progress_bar, managed_monitor_container_log_task(
-                container=container,
-                progress_regexp=image_labels.get_progress_regexp(),
-                service_key=self.task_parameters.image,
-                service_version=self.task_parameters.tag,
-                task_publishers=self.task_publishers,
-                integration_version=image_labels.get_integration_version(),
-                task_volumes=task_volumes,
-                log_file_url=self.log_file_url,
-                log_publishing_cb=self._publish_sidecar_log,
-                s3_settings=self.s3_settings,
-                progress_bar=processing_progress_bar,
+            async with (
+                managed_container(
+                    docker_client,
+                    config,
+                    name=f"{self.task_parameters.image.split(sep='/')[-1]}_{run_id}",
+                ) as container,
+                progress_bar.sub_progress(
+                    100, description="processing"
+                ) as processing_progress_bar,
+                managed_monitor_container_log_task(
+                    container=container,
+                    progress_regexp=image_labels.get_progress_regexp(),
+                    service_key=self.task_parameters.image,
+                    service_version=self.task_parameters.tag,
+                    task_publishers=self.task_publishers,
+                    integration_version=image_labels.get_integration_version(),
+                    task_volumes=task_volumes,
+                    log_file_url=self.log_file_url,
+                    log_publishing_cb=self._publish_sidecar_log,
+                    s3_settings=self.s3_settings,
+                    progress_bar=processing_progress_bar,
+                ),
             ):
                 await container.start()
                 await self._publish_sidecar_log(

services/dask-sidecar/src/simcore_service_dask_sidecar/computational_sidecar/docker_utils.py

@@ -188,7 +188,7 @@ async def _parse_and_publish_logs(
     if progress_value is not None:
         await progress_bar.set_(round(progress_value * 100.0))
 
-    task_publishers.publish_logs(
+    await task_publishers.publish_logs(
         message=log_line, log_level=guess_message_log_level(log_line)
     )
 

services/dask-sidecar/src/simcore_service_dask_sidecar/dask_utils.py

@@ -138,7 +138,7 @@ async def cancel_task(task_name: str) -> None:
         if task := next(
             (t for t in asyncio.all_tasks() if t.get_name() == task_name), None
         ):
-            task_publishers.publish_logs(
+            await task_publishers.publish_logs(
                 message="[sidecar] cancelling task...", log_level=logging.INFO
             )
             task.cancel()
@@ -158,7 +158,7 @@ async def periodicaly_check_if_aborted(task_name: str) -> None:
 
         yield
     except asyncio.CancelledError as exc:
-        task_publishers.publish_logs(
+        await task_publishers.publish_logs(
             message="[sidecar] task run was aborted", log_level=logging.INFO
         )
 

services/dask-sidecar/src/simcore_service_dask_sidecar/scheduler.py

@@ -4,8 +4,8 @@
 from servicelib.logging_utils import log_context
 
 from ._meta import print_dask_scheduler_banner
-from .app_utils import setup_app_logging
 from .settings import ApplicationSettings
+from .utils.logs import setup_app_logging
 
 _logger = logging.getLogger(__name__)
 

services/dask-sidecar/src/simcore_service_dask_sidecar/app_utils.py renamed to services/dask-sidecar/src/simcore_service_dask_sidecar/utils/logs.py

@@ -2,7 +2,7 @@
 
 from servicelib.logging_utils import config_all_loggers
 
-from .settings import ApplicationSettings
+from ..settings import ApplicationSettings
 
 
 def setup_app_logging(settings: ApplicationSettings) -> None:

services/dask-sidecar/src/simcore_service_dask_sidecar/worker.py

@@ -15,11 +15,11 @@
 from settings_library.s3 import S3Settings
 
 from ._meta import print_dask_sidecar_banner
-from .app_utils import setup_app_logging
 from .computational_sidecar.core import ComputationalSidecar
 from .dask_utils import TaskPublisher, get_current_task_resources, monitor_task_abortion
 from .rabbitmq_plugin import RabbitMQPlugin
 from .settings import ApplicationSettings
+from .utils.logs import setup_app_logging
 
 _logger = logging.getLogger(__name__)