fix timeout

Author: sanderegg

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

@@ -10,6 +10,7 @@
 
 import aiofiles
 import aiofiles.tempfile
+import aiohttp
 import arrow
 from aiodocker import Docker, DockerError
 from aiodocker.containers import DockerContainer
@@ -268,42 +269,57 @@ async def _parse_container_docker_logs(
     with log_context(
         logger, logging.DEBUG, "started monitoring of >=1.0 service - using docker logs"
     ):
-        async with aiofiles.tempfile.TemporaryDirectory() as tmp_dir:
-            log_file_path = (
-                Path(tmp_dir)
-                / f"{service_key.split(sep='/')[-1]}_{service_version}.logs"
+        assert isinstance(container.docker.connector, aiohttp.UnixConnector)  # nosec
+        async with Docker(
+            session=aiohttp.ClientSession(
+                connector=aiohttp.UnixConnector(container.docker.connector.path),
+                timeout=aiohttp.ClientTimeout(total=_AIODOCKER_LOGS_TIMEOUT_S),
             )
-            log_file_path.parent.mkdir(parents=True, exist_ok=True)
-            async with aiofiles.open(log_file_path, mode="wb+") as log_fp:
-                async for log_line in cast(
-                    AsyncGenerator[str, None],
-                    container.log(
-                        stdout=True,
-                        stderr=True,
-                        follow=True,
-                        timestamps=True,
-                        timeout=_AIODOCKER_LOGS_TIMEOUT_S,
-                    ),
-                ):
-                    log_msg_without_timestamp = log_line.split(" ", maxsplit=1)[1]
-                    logger.info(
-                        "[%s]: %s",
-                        f"{service_key}:{service_version} - {container.id}{container_name}",
-                        log_msg_without_timestamp,
-                    )
-                    await log_fp.write(log_line.encode("utf-8"))
-                    # NOTE: here we remove the timestamp, only needed for the file
-                    await _parse_and_publish_logs(
-                        log_msg_without_timestamp,
-                        task_publishers=task_publishers,
-                        progress_regexp=progress_regexp,
-                        progress_bar=progress_bar,
-                    )
-
-            # copy the log file to the log_file_url
-            await push_file_to_remote(
-                log_file_path, log_file_url, log_publishing_cb, s3_settings
+        ) as docker_client_for_logs:
+            # NOTE: this is a workaround for aiodocker not being able to get the container
+            # logs when the container is not running
+            container_for_long_running_logs = (
+                await docker_client_for_logs.containers.get(container.id)
             )
+            # NOTE: this is a workaround for aiodocker not being able to get the container
+            # logs when the container is not running
+            await container.show()
+            await container_for_long_running_logs.show()
+            async with aiofiles.tempfile.TemporaryDirectory() as tmp_dir:
+                log_file_path = (
+                    Path(tmp_dir)
+                    / f"{service_key.split(sep='/')[-1]}_{service_version}.logs"
+                )
+                log_file_path.parent.mkdir(parents=True, exist_ok=True)
+                async with aiofiles.open(log_file_path, mode="wb+") as log_fp:
+                    async for log_line in cast(
+                        AsyncGenerator[str, None],
+                        container_for_long_running_logs.log(
+                            stdout=True,
+                            stderr=True,
+                            follow=True,
+                            timestamps=True,
+                        ),
+                    ):
+                        log_msg_without_timestamp = log_line.split(" ", maxsplit=1)[1]
+                        logger.info(
+                            "[%s]: %s",
+                            f"{service_key}:{service_version} - {container_for_long_running_logs.id}{container_name}",
+                            log_msg_without_timestamp,
+                        )
+                        await log_fp.write(log_line.encode("utf-8"))
+                        # NOTE: here we remove the timestamp, only needed for the file
+                        await _parse_and_publish_logs(
+                            log_msg_without_timestamp,
+                            task_publishers=task_publishers,
+                            progress_regexp=progress_regexp,
+                            progress_bar=progress_bar,
+                        )
+
+                # copy the log file to the log_file_url
+                await push_file_to_remote(
+                    log_file_path, log_file_url, log_publishing_cb, s3_settings
+                )
 
 
 async def _monitor_container_logs(  # noqa: PLR0913 # pylint: disable=too-many-arguments
@@ -325,6 +341,7 @@ async def _monitor_container_logs(  # noqa: PLR0913 # pylint: disable=too-many-a
     Services above are not creating a file and use the usual docker logging. These logs
     are retrieved using the usual cli 'docker logs CONTAINERID'
     """
+
     with log_catch(logger, reraise=False):
         container_info = await container.show()
         container_name = container_info.get("Name", "undefined")

services/dask-sidecar/tests/unit/test_tasks.py

@@ -208,13 +208,13 @@ def sleeper_task(
             "the_input_43": 15.0,
             "the_bool_input_54": False,
             **{
-                f"some_file_input_{index+1}": FileUrl(url=file)
+                f"some_file_input_{index + 1}": FileUrl(url=file)
                 for index, file in enumerate(list_of_files)
             },
             **{
-                f"some_file_input_with_mapping{index+1}": FileUrl(
+                f"some_file_input_with_mapping{index + 1}": FileUrl(
                     url=file,
-                    file_mapping=f"{index+1}/some_file_input_{index+1}",
+                    file_mapping=f"{index + 1}/some_file_input_{index + 1}",
                 )
                 for index, file in enumerate(list_of_files)
             },
@@ -241,7 +241,7 @@ def sleeper_task(
     )
     list_of_bash_commands += _bash_check_env_exist(
         variable_name="SIMCORE_NANO_CPUS_LIMIT",
-        variable_value=f"{int(_DEFAULT_MAX_RESOURCES['CPU']*1e9)}",
+        variable_value=f"{int(_DEFAULT_MAX_RESOURCES['CPU'] * 1e9)}",
     )
     list_of_bash_commands += _bash_check_env_exist(
         variable_name="SIMCORE_MEMORY_BYTES_LIMIT",
@@ -268,7 +268,7 @@ def sleeper_task(
         f"echo '{faker.text(max_nb_chars=17216)}'",
         f"(test -f ${{INPUT_FOLDER}}/{input_json_file_name} || (echo ${{INPUT_FOLDER}}/{input_json_file_name} file does not exists && exit 1))",
         f"echo $(cat ${{INPUT_FOLDER}}/{input_json_file_name})",
-        f"sleep {randint(1,4)}",  # noqa: S311
+        f"sleep {randint(1, 4)}",  # noqa: S311
     ]
 
     # defines the expected outputs
@@ -435,6 +435,13 @@ def sleeper_task_unexpected_output(
     return sleeper_task
 
 
+@pytest.fixture
+def hugely_talkative_ubuntu_task(
+    sidecar_task: Callable[..., ServiceExampleParam],
+) -> ServiceExampleParam:
+    return sidecar_task(command=["/bin/bash", "-c", "some stupid failing command"])
+
+
 @pytest.fixture()
 def caplog_info_level(
     caplog: pytest.LogCaptureFixture,
@@ -709,3 +716,47 @@ def test_running_service_that_generates_unexpected_data_raises_exception(
         run_computational_sidecar(
             **sleeper_task_unexpected_output.sidecar_params(),
         )
+
+
+@pytest.mark.parametrize(
+    "integration_version, boot_mode", [("1.0.0", BootMode.CPU)], indirect=True
+)
+def test_delayed_logging_with_small_timeout_raises_exception(
+    caplog: pytest.LogCaptureFixture,
+    app_environment: EnvVarsDict,
+    dask_subsystem_mock: dict[str, mock.Mock],
+    sidecar_task: Callable[..., ServiceExampleParam],
+    mocked_get_image_labels: mock.Mock,
+    mocker: MockerFixture,
+):
+    """https://github.com/aio-libs/aiodocker/issues/901"""
+    # Mock the timeout with a very small value
+    mocker.patch(
+        "simcore_service_dask_sidecar.computational_sidecar.docker_utils._AIODOCKER_LOGS_TIMEOUT_S",
+        0.5,  # Small timeout that should cause failure
+    )
+
+    # Configure the task to sleep first and then generate logs
+    waiting_task = sidecar_task(
+        command=[
+            "/bin/bash",
+            "-c",
+            'echo "Starting task"; sleep 5; echo "After sleep"',
+        ]
+    )
+
+    # Execute the task and expect a timeout exception in the logs
+    with caplog.at_level(logging.ERROR, logger="simcore_service_dask_sidecar"):
+        run_computational_sidecar(**waiting_task.sidecar_params())
+        assert len(caplog.records) == 1
+        record = caplog.records[0]
+        assert record.exc_info
+        assert isinstance(record.exc_info[1], TimeoutError)
+    caplog.clear()
+    mocker.patch(
+        "simcore_service_dask_sidecar.computational_sidecar.docker_utils._AIODOCKER_LOGS_TIMEOUT_S",
+        10,  # larger timeout to avoid issues
+    )
+    with caplog.at_level(logging.ERROR, logger="simcore_service_dask_sidecar"):
+        run_computational_sidecar(**waiting_task.sidecar_params())
+        assert len(caplog.records) == 0