fix failing to close aiodocker connection (#1765)

Author: sanderegg

services/director/tests/fixtures/fake_services.py

@@ -2,6 +2,7 @@
 # pylint:disable=redefined-outer-name
 
 
+import asyncio
 import json
 import logging
 import random
@@ -10,13 +11,12 @@
 
 import pytest
 import requests
+from aiodocker import utils
 from aiodocker.docker import Docker
 from aiodocker.exceptions import DockerError
 
 _logger = logging.getLogger(__name__)
 
-import asyncio
-
 
 @pytest.fixture(scope="function")
 def push_services(loop, docker_registry, tmpdir):
@@ -98,7 +98,7 @@ async def _build_push_image(
     *,
     bad_json_format=False,
 ):  # pylint: disable=R0913
-    docker = Docker()
+
     # crate image
     service_description = _create_service_description(service_type, name, tag)
     docker_labels = _create_docker_labels(service_description, bad_json_format)
@@ -157,7 +157,9 @@ async def _build_push_image(
     await _create_base_image(docker_labels, image_tag)
 
     # push image to registry
+    docker = Docker()
     await docker.images.push(image_tag)
+    await docker.close()
     # remove image from host
     # docker.images.remove(image_tag)
     return {
@@ -190,9 +192,6 @@ def _clean_registry(registry_url, list_of_images):
         response = requests.delete(url, headers=request_headers)
 
 
-from aiodocker import utils
-
-
 async def _create_base_image(labels, tag):
     dockerfile = """
 FROM alpine
@@ -206,6 +205,7 @@ async def _create_base_image(labels, tag):
     base_docker_image = await docker.images.build(
         fileobj=tar_obj, encoding="gzip", rm=True, labels=labels, tag=tag
     )
+    await docker.close()
     return base_docker_image[0]
 
 

services/sidecar/src/simcore_service_sidecar/executor.py

@@ -6,16 +6,18 @@
 from pathlib import Path
 from typing import Dict, Optional
 
+import aiopg
 import attr
+from aiodocker import Docker
+from aiodocker.containers import DockerContainer
+from aiodocker.exceptions import DockerContainerError, DockerError
+from packaging import version
+from tenacity import retry, stop_after_attempt
 
-import aiodocker
-import aiopg
 from celery.utils.log import get_task_logger
-from packaging import version
 from servicelib.utils import fire_and_forget_task, logged_gather
 from simcore_sdk import node_data, node_ports
 from simcore_sdk.node_ports.dbmanager import DBManager
-from tenacity import retry, stop_after_attempt
 
 from . import config, exceptions
 from .boot_mode import get_boot_mode
@@ -98,7 +100,7 @@ async def run(self):
         except asyncio.CancelledError:
             await self._post_messages(LogType.LOG, "[sidecar]...task cancelled")
             raise
-        except (aiodocker.exceptions.DockerError, exceptions.SidecarException) as exc:
+        except (DockerError, exceptions.SidecarException) as exc:
             await self._post_messages(
                 LogType.LOG, f"[sidecar]...task failed: {str(exc)}"
             )
@@ -215,40 +217,37 @@ async def _pull_image(self):
             config.DOCKER_PASSWORD,
         )
         try:
-            docker_client: aiodocker.Docker = aiodocker.Docker()
-            await self._post_messages(
-                LogType.LOG,
-                f"[sidecar]Pulling {self.task.image['name']}:{self.task.image['tag']}...",
-            )
-            await docker_client.images.pull(
-                docker_image,
-                auth={
-                    "username": config.DOCKER_USER,
-                    "password": config.DOCKER_PASSWORD,
-                },
-            )
-
-            # get integration version
-            image_cfg = await docker_client.images.inspect(docker_image)
-            # NOTE: old services did not have that label
-            if "io.simcore.integration-version" in image_cfg["Config"]["Labels"]:
-                self.integration_version = version.parse(
-                    json.loads(
-                        image_cfg["Config"]["Labels"]["io.simcore.integration-version"]
-                    )["integration-version"]
+            async with Docker() as docker_client:
+                await self._post_messages(
+                    LogType.LOG,
+                    f"[sidecar]Pulling {self.task.image['name']}:{self.task.image['tag']}...",
+                )
+                await docker_client.images.pull(
+                    docker_image,
+                    auth={
+                        "username": config.DOCKER_USER,
+                        "password": config.DOCKER_PASSWORD,
+                    },
                 )
 
-        except aiodocker.exceptions.DockerError:
+                # get integration version
+                image_cfg = await docker_client.images.inspect(docker_image)
+                # NOTE: old services did not have that label
+                if "io.simcore.integration-version" in image_cfg["Config"]["Labels"]:
+                    self.integration_version = version.parse(
+                        json.loads(
+                            image_cfg["Config"]["Labels"][
+                                "io.simcore.integration-version"
+                            ]
+                        )["integration-version"]
+                    )
+
+        except DockerError:
             msg = f"Failed to pull image '{docker_image}'"
             log.exception(msg)
             raise
 
-    async def _run_container(self):
-        # pylint: disable=too-many-statements, too-many-branches
-        start_time = time.perf_counter()
-        container = None
-        docker_image = f"{config.DOCKER_REGISTRY}/{self.task.image['name']}:{self.task.image['tag']}"
-
+    async def _create_container_config(self, docker_image: str) -> Dict:
         # NOTE: Env/Binds for log folder is only necessary for integraion "0"
         env_vars = [
             f"{name.upper()}_FOLDER=/{name}/{self.task.job_id}"
@@ -293,89 +292,105 @@ async def _run_container(self):
                 ],
             },
         }
-        log.debug(
-            "Running image %s with config %s", docker_image, docker_container_config
+        return docker_container_config
+
+    async def _start_monitoring_container(
+        self, container: DockerContainer
+    ) -> asyncio.Future:
+        log_file = self.shared_folders.log_folder / "log.dat"
+        if self.integration_version == version.parse("0.0.0"):
+            # touch output file, so it's ready for the container (v0)
+            log_file.touch()
+
+            log_processor_task = fire_and_forget_task(
+                monitor_logs_task(log_file, self._post_messages)
+            )
+            return log_processor_task
+        log_processor_task = fire_and_forget_task(
+            monitor_logs_task(container, self._post_messages, log_file)
         )
+        return log_processor_task
+
+    async def _run_container(self):
+        start_time = time.perf_counter()
+        docker_image = f"{config.DOCKER_REGISTRY}/{self.task.image['name']}:{self.task.image['tag']}"
+        container_config = await self._create_container_config(docker_image)
+
         # volume paths for car container (w/o prefix)
         result = "FAILURE"
         log_processor_task = None
         try:
-            docker_client: aiodocker.Docker = aiodocker.Docker()
-            await self._post_messages(
-                LogType.LOG,
-                f"[sidecar]Running {self.task.image['name']}:{self.task.image['tag']}...",
-            )
-            container = await docker_client.containers.create(
-                config=docker_container_config
-            )
-            # start monitoring logs
-            log_file = self.shared_folders.log_folder / "log.dat"
-            if self.integration_version == version.parse("0.0.0"):
-                # touch output file, so it's ready for the container (v0)
-                log_file.touch()
-
-                log_processor_task = fire_and_forget_task(
-                    monitor_logs_task(log_file, self._post_messages)
+            async with Docker() as docker_client:
+                await self._post_messages(
+                    LogType.LOG,
+                    f"[sidecar]Running {self.task.image['name']}:{self.task.image['tag']}...",
                 )
-            else:
-                log_processor_task = fire_and_forget_task(
-                    monitor_logs_task(container, self._post_messages, log_file)
+                container = await docker_client.containers.create(
+                    config=container_config
+                )
+                log_processor_task = await self._start_monitoring_container(container)
+                # start the container
+                await container.start()
+                # indicate container is started
+                await self.rabbit_mq.post_instrumentation_message(
+                    {
+                        "metrics": "service_started",
+                        "user_id": self.user_id,
+                        "project_id": self.task.project_id,
+                        "service_uuid": self.task.node_id,
+                        "service_type": "COMPUTATIONAL",
+                        "service_key": self.task.image["name"],
+                        "service_tag": self.task.image["tag"],
+                    }
                 )
-            # start the container
-            await container.start()
-            # indicate container is started
-            await self.rabbit_mq.post_instrumentation_message(
-                {
-                    "metrics": "service_started",
-                    "user_id": self.user_id,
-                    "project_id": self.task.project_id,
-                    "service_uuid": self.task.node_id,
-                    "service_type": "COMPUTATIONAL",
-                    "service_key": self.task.image["name"],
-                    "service_tag": self.task.image["tag"],
-                }
-            )
 
-            # wait until the container finished, either success or fail or timeout
-            container_data = await container.show()
-            while container_data["State"]["Running"]:
-                await asyncio.sleep(2)
-                # reload container data
+                # wait until the container finished, either success or fail or timeout
                 container_data = await container.show()
-                if (
-                    (time.perf_counter() - start_time) > config.SERVICES_TIMEOUT_SECONDS
-                    and config.SERVICES_TIMEOUT_SECONDS > 0
-                ):
-                    log.error(
-                        "Running container timed-out after %ss and will be stopped now\nlogs: %s",
-                        config.SERVICES_TIMEOUT_SECONDS,
-                        container.log(stdout=True, stderr=True),
+                TIME_TO_NEXT_PERDIODIC_CHECK_SECS = 2
+                while container_data["State"]["Running"]:
+                    await asyncio.sleep(TIME_TO_NEXT_PERDIODIC_CHECK_SECS)
+                    # reload container data
+                    container_data = await container.show()
+                    if (
+                        (time.perf_counter() - start_time)
+                        > config.SERVICES_TIMEOUT_SECONDS
+                        and config.SERVICES_TIMEOUT_SECONDS > 0
+                    ):
+                        log.error(
+                            "Running container timed-out after %ss and will be stopped now\nlogs: %s",
+                            config.SERVICES_TIMEOUT_SECONDS,
+                            container.log(stdout=True, stderr=True),
+                        )
+                        await container.stop()
+                        break
+
+                # reload container data to check the error code with latest info
+                container_data = await container.show()
+                if container_data["State"]["ExitCode"] > 0:
+                    exc = exceptions.SidecarException(
+                        f"{docker_image} completed with error code {container_data['State']['ExitCode']}:\n {container_data['State']['Error']}\n:Last logs:\n{container.logs(stdout=True, stderr=True, tail=10)}"
                     )
-                    await container.stop()
-                    break
-
-            # reload container data to check the error code with latest info
-            container_data = await container.show()
-            if container_data["State"]["ExitCode"] > 0:
-                raise exceptions.SidecarException(
-                    f"{docker_image} completed with error code {container_data['State']['ExitCode']}:\n {container_data['State']['Error']}\n:Last logs:\n{container.logs(stdout=True, stderr=True, tail=10)}"
-                )
-            # ensure progress 1.0 is sent
-            await self._post_messages(LogType.PROGRESS, "1.0")
-            result = "SUCCESS"
-            log.info("%s completed with successfully!", docker_image)
-        except aiodocker.exceptions.DockerContainerError:
+                    # clean up the container
+                    await container.delete(force=True)
+                    raise exc
+                # clean up the container
+                await container.delete(force=True)
+                # ensure progress 1.0 is sent
+                await self._post_messages(LogType.PROGRESS, "1.0")
+                result = "SUCCESS"
+                log.info("%s completed with successfully!", docker_image)
+        except DockerContainerError:
             log.exception(
                 "Error while running %s with parameters %s",
                 docker_image,
-                docker_container_config,
+                container_config,
             )
             raise
-        except aiodocker.exceptions.DockerError:
+        except DockerError:
             log.exception(
                 "Unknown error while trying to run %s with parameters %s",
                 docker_image,
-                docker_container_config,
+                container_config,
             )
             raise
         except asyncio.CancelledError:
@@ -385,12 +400,10 @@ async def _run_container(self):
         finally:
             stop_time = time.perf_counter()
             log.info("Running %s took %sseconds", docker_image, stop_time - start_time)
-            if container:
-                # clean up the container
-                await container.delete(force=True)
             # stop monitoring logs now
             if log_processor_task:
                 log_processor_task.cancel()
+                await log_processor_task
             # instrumentation
             await self.rabbit_mq.post_instrumentation_message(
                 {
@@ -404,8 +417,6 @@ async def _run_container(self):
                     "result": result,
                 }
             )
-            if log_processor_task:
-                await log_processor_task
 
     async def _process_task_output(self):
         """There will be some files in the /output

services/sidecar/src/simcore_service_sidecar/utils.py

@@ -141,9 +141,9 @@ def assemble_celery_app(task_default_queue: str, rabbit_config: RabbitConfig) ->
 
 async def get_volume_mount_point(volume_name: str) -> str:
     try:
-        docker_client: aiodocker.Docker = aiodocker.Docker()
-        volume_attributes = await DockerVolume(docker_client, volume_name).show()
-        return volume_attributes["Mountpoint"]
+        async with aiodocker.Docker() as docker_client:
+            volume_attributes = await DockerVolume(docker_client, volume_name).show()
+            return volume_attributes["Mountpoint"]
 
     except aiodocker.exceptions.DockerError as err:
         raise SidecarException(

services/sidecar/tests/integration/test_sidecar.py

@@ -371,7 +371,7 @@ async def rabbit_message_handler(message: aio_pika.IncomingMessage):
     for key in pipeline_cfg:
         dag.extend(pipeline_cfg[key]["next"])
     assert next_task_nodes == dag
-
+    await asyncio.sleep(5)  # wait a little bit for logs to come in
     _assert_incoming_data_logs(
         list(pipeline_cfg.keys()),
         incoming_data,