Remove stale KVM children and keep IOLoop responsive during connect

nojava_ipmi_kvm/kvm.py

@@ -16,6 +16,7 @@
 
 from .utils import generate_temp_password
 from .config import config, HostConfig, HTML5HostConfig, JavaHostConfig
+from .stale_children import cleanup_stale_kvm_children
 from ._version import __version__
 
 logger = logging.getLogger(__name__)
@@ -130,9 +131,18 @@ def html5_endpoint(self):
 
 
 def log_factory(additional_logging):
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = None
+
     def log(msg, *args, **kwargs):
         logger.info(msg, *args, **kwargs)
-        if additional_logging is not None:
+        if additional_logging is None:
+            return
+        if loop is not None:
+            loop.call_soon_threadsafe(additional_logging, msg, *args, **kwargs)
+        else:
             additional_logging(msg, *args, **kwargs)
 
     return log
@@ -145,6 +155,12 @@ def add_sudo_if_configured(command_list):
     return command_list
 
 
+async def _run_blocking(func):
+    # type: (Callable[[], Any]) -> Any
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(None, func)
+
+
 async def check_webserver(log, url):
     # type: (Callable, Text) -> None
     log("Check if '%s' is reachable...", url)
@@ -161,19 +177,22 @@ async def check_docker(log, subprocess_output):
     # type: (Callable, Optional[int]) -> None
     if not is_command_available("docker"):
         raise DockerNotInstalledError("Could not find the `docker` command. Please install Docker first.")
-    if (
-        subprocess.call(add_sudo_if_configured(["docker", "ps"]), stdout=subprocess_output, stderr=subprocess_output)
-        != 0
-    ):
+
+    def docker_ps():
+        # type: () -> int
+        return subprocess.call(
+            add_sudo_if_configured(["docker", "ps"]), stdout=subprocess_output, stderr=subprocess_output
+        )
+
+    if await _run_blocking(docker_ps) != 0:
         if running_macos():
-            subprocess.check_call(["open", "-g", "-a", "Docker"])
+
+            def open_docker():
+                subprocess.check_call(["open", "-g", "-a", "Docker"])
+
+            await _run_blocking(open_docker)
             log("Waiting for the Docker engine to be ready...")
-            while (
-                subprocess.call(
-                    add_sudo_if_configured(["docker", "ps"]), stdout=subprocess_output, stderr=subprocess_output
-                )
-                != 0
-            ):
+            while await _run_blocking(docker_ps) != 0:
                 await asyncio.sleep(1)
         else:
             raise DockerNotCallableError(
@@ -182,6 +201,18 @@ async def check_docker(log, subprocess_output):
             )
 
 
+def docker_terminated_message(return_code, docker_port=None):
+    # type: (int, Optional[int]) -> Text
+    message = "Docker terminated with return code {}.".format(return_code)
+    if return_code == 125:
+        port_hint = docker_port if docker_port is not None else "N"
+        message += (
+            " Stale KVM container may be holding port {}. "
+            "Retry or remove nojava-ipmi-kvmrc-* manually.".format(port_hint)
+        )
+    return message
+
+
 def create_extra_args(host_config):
     # type: (HostConfig) -> List
     extra_args = [
@@ -287,6 +318,14 @@ async def start_kvm_container(
     await check_webserver(log, "http://{}/".format(host_config.full_hostname))
     await check_docker(log, subprocess_output)
 
+    port_start = int(os.environ.get("WEB_PORT_START", 8800))
+    port_end = int(os.environ.get("WEB_PORT_END", 8900))
+
+    def run_cleanup():
+        cleanup_stale_kvm_children(port_start, port_end, log=log)
+
+    await _run_blocking(run_cleanup)
+
     # TODO: pass variables as `extra_args` (?)
     DOCKER_CONTAINER_NAME = "nojava-ipmi-kvmrc-{}".format(uuid.uuid4())
 
@@ -300,25 +339,30 @@ async def start_kvm_container(
         )
 
     log("Starting the Docker container...")
-    docker_process = subprocess.Popen(
-        add_sudo_if_configured(
-            ["docker", "run", "-i", "-v", "/etc/hosts:/etc/hosts:ro", "--rm", "--name", DOCKER_CONTAINER_NAME]
+
+    def launch_docker():
+        # type: () -> subprocess.Popen
+        docker_process = subprocess.Popen(
+            add_sudo_if_configured(
+                ["docker", "run", "-i", "-v", "/etc/hosts:/etc/hosts:ro", "--rm", "--name", DOCKER_CONTAINER_NAME]
+            )
+            + environment_variables
+            + (["-P"] if docker_port is None else ["-p", "{}:8080".format(docker_port)])
+            + [docker_image]
+            + extra_args,
+            stdin=subprocess.PIPE,
+            stdout=subprocess_output,
+            stderr=subprocess_output,
         )
-        + environment_variables
-        + (["-P"] if docker_port is None else ["-p", "{}:8080".format(docker_port)])
-        + [docker_image]
-        + extra_args,
-        stdin=subprocess.PIPE,
-        stdout=subprocess_output,
-        stderr=subprocess_output,
-    )
-    if docker_process.stdin is not None:
-        docker_process.stdin.write("{}\n".format(stdin).encode("utf-8"))
-        docker_process.stdin.flush()
-        docker_process.stdin.close()
-    else:
-        # This case cannot happen (`if` is used to satisfy mypy)
-        raise IOError("Something strange happened: Docker stdin not available.")
+        if docker_process.stdin is not None:
+            docker_process.stdin.write("{}\n".format(stdin).encode("utf-8"))
+            docker_process.stdin.flush()
+            docker_process.stdin.close()
+        else:
+            raise IOError("Something strange happened: Docker stdin not available.")
+        return docker_process
+
+    docker_process = await _run_blocking(launch_docker)
 
     def terminate_docker():
         # type: () -> None
@@ -333,18 +377,21 @@ def terminate_docker():
 
     while True:
         try:
-            if docker_process.poll() is not None:
-                raise DockerTerminatedError("Docker terminated with return code {}.".format(docker_process.returncode))
-            web_port = int(
-                subprocess.check_output(
+            poll_result = await _run_blocking(docker_process.poll)
+            if poll_result is not None:
+                raise DockerTerminatedError(docker_terminated_message(docker_process.returncode, docker_port))
+
+            def read_web_port():
+                # type: () -> int
+                port_output = subprocess.check_output(
                     add_sudo_if_configured(["docker", "port", DOCKER_CONTAINER_NAME]), stderr=subprocess_output
                 )
-                .strip()
-                .split(b"\n")[0].split(b":")[1]
-            )
+                return int(port_output.strip().split(b"\n")[0].split(b":")[1])
+
+            web_port = await _run_blocking(read_web_port)
             break
         except (IndexError, ValueError):
-            terminate_docker()
+            await _run_blocking(terminate_docker)
             raise DockerPortNotReadableError("Cannot read the VNC web port.")
         except subprocess.CalledProcessError:
             await asyncio.sleep(1)
@@ -365,19 +412,17 @@ def get():
             response.raise_for_status()
             break
         except (requests.ConnectionError, requests.HTTPError):
-            if docker_process.poll() is not None:
+            poll_result = await _run_blocking(docker_process.poll)
+            if poll_result is not None:
                 if not host_config.skip_login:
                     raise DockerTerminatedError(
-                        "Docker terminated with return code {}. Maybe you entered a wrong password?".format(
-                            docker_process.returncode
-                        )
+                        docker_terminated_message(docker_process.returncode, docker_port)
+                        + " Maybe you entered a wrong password?"
                     )
                 else:
                     raise DockerTerminatedError(
-                        (
-                            "Docker terminated with return code {}."
-                            + " Maybe you configured a wrong download endpoint or need a login?"
-                        ).format(docker_process.returncode)
+                        docker_terminated_message(docker_process.returncode, docker_port)
+                        + " Maybe you configured a wrong download endpoint or need a login?"
                     )
             await asyncio.sleep(1)
 

nojava_ipmi_kvm/stale_children.py

@@ -0,0 +1,101 @@
+import logging
+import subprocess
+
+try:
+    from typing import Any, Callable, Optional  # noqa: F401  # pylint: disable=unused-import
+except ImportError:
+    pass
+
+from .config import config
+
+logger = logging.getLogger(__name__)
+
+CHILD_NAME_PREFIX = "nojava-ipmi-kvmrc-"
+
+
+def _add_sudo_if_configured(command_list):
+    if config.run_docker_with_sudo:
+        command_list.insert(0, "sudo")
+    return command_list
+
+
+def _port_in_range(port, port_start, port_end):
+    if port is None:
+        return False
+    try:
+        port_value = int(port)
+    except (TypeError, ValueError):
+        return False
+    return port_start <= port_value < port_end
+
+
+def cleanup_stale_kvm_children(port_start, port_end, log=None):
+    # type: (int, int, Optional[Callable[..., None]]) -> None
+    log_func = log if log is not None else logger.info
+
+    list_result = subprocess.run(
+        _add_sudo_if_configured(
+            ["docker", "ps", "-aq", "--filter", "name={}".format(CHILD_NAME_PREFIX)]
+        ),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        universal_newlines=True,
+        check=False,
+    )
+
+    for container_id in list_result.stdout.split():
+        container_id = container_id.strip()
+        if not container_id:
+            continue
+
+        state_result = subprocess.run(
+            _add_sudo_if_configured(["docker", "inspect", "-f", "{{.State.Status}}", container_id]),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            universal_newlines=True,
+            check=False,
+        )
+        state = state_result.stdout.strip()
+
+        name_result = subprocess.run(
+            _add_sudo_if_configured(["docker", "inspect", "-f", "{{.Name}}", container_id]),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            universal_newlines=True,
+            check=False,
+        )
+        name = name_result.stdout.strip().lstrip("/")
+
+        if state in ("exited", "dead"):
+            remove_result = subprocess.run(
+                _add_sudo_if_configured(["docker", "rm", "-f", container_id]),
+                stderr=subprocess.DEVNULL,
+                check=False,
+            )
+            if remove_result.returncode == 0:
+                log_func("Removed exited child {}".format(name))
+            continue
+
+        port_result = subprocess.run(
+            _add_sudo_if_configured(["docker", "port", container_id, "8080/tcp"]),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            universal_newlines=True,
+            check=False,
+        )
+        host_port = None
+        if port_result.returncode == 0 and port_result.stdout.strip():
+            host_port = port_result.stdout.strip().splitlines()[0].rsplit(":", 1)[-1]
+
+        if _port_in_range(host_port, port_start, port_end):
+            remove_result = subprocess.run(
+                _add_sudo_if_configured(["docker", "rm", "-f", container_id]),
+                stderr=subprocess.DEVNULL,
+                check=False,
+            )
+            if remove_result.returncode == 0:
+                log_func(
+                    "Removed stale child {} (port {}, range {}-{})".format(
+                        name, host_port, port_start, port_end - 1
+                    )
+                )