Source code for galaxy.tool_util.deps.docker_util

"""Utilities for building up Docker commands...

...using common defaults and configuration mechanisms.
"""

import os
import shlex
import sys
from typing import (
    List,
    Optional,
    TYPE_CHECKING,
    Union,
)

if TYPE_CHECKING:
    from .container_volumes import DockerVolume

from galaxy.util.commands import argv_to_str

DEFAULT_DOCKER_COMMAND = "docker"
DEFAULT_SUDO = False
DEFAULT_SUDO_COMMAND = "sudo"
DEFAULT_HOST = None
DEFAULT_VOLUME_MOUNT_TYPE = "rw"
DEFAULT_WORKING_DIRECTORY = None
DEFAULT_NET = None
DEFAULT_MEMORY = None
DEFAULT_VOLUMES_FROM = None
DEFAULT_AUTO_REMOVE = True
DEFAULT_SET_USER = None if sys.platform == "darwin" else "$UID"
DEFAULT_RUN_EXTRA_ARGUMENTS = None


[docs]def kill_command(container: str, signal: Optional[str] = None, **kwds) -> List[str]:
    args = (["-s", signal] if signal else []) + [container]
    return command_list("kill", args, **kwds)


[docs]def logs_command(container: str, **kwds) -> List[str]:
    return command_list("logs", [container], **kwds)


[docs]def build_command(image: str, docker_build_path: str, **kwds) -> List[str]:
    if os.path.isfile(docker_build_path):
        docker_build_path = os.path.dirname(os.path.abspath(docker_build_path))
    return command_list("build", ["-t", image, docker_build_path], **kwds)


[docs]def build_save_image_command(image: str, destination: str, **kwds) -> List[str]:
    return command_list("save", ["-o", destination, image], **kwds)


[docs]def build_pull_command(tag: str, **kwds) -> List[str]:
    return command_list("pull", [tag], **kwds)


[docs]def build_docker_cache_command(image: str, **kwds) -> str:
    assert kwds.get("to_str", True)
    inspect_image_command = command_shell("inspect", [image], **kwds)
    pull_image_command = command_shell("pull", [image], **kwds)
    cache_command = f"{inspect_image_command} > /dev/null 2>&1\n[ $? -ne 0 ] && {pull_image_command} > /dev/null 2>&1\n"
    return cache_command


[docs]def build_docker_images_command(truncate=True, **kwds) -> Union[str, List[str]]:
    args = ["--no-trunc"] if not truncate else []
    return command_shell("images", args, **kwds)


[docs]def build_docker_load_command(**kwds) -> Union[str, List[str]]:
    return command_shell("load", [])


[docs]def build_docker_simple_command(
    command: str,
    docker_cmd: str = DEFAULT_DOCKER_COMMAND,
    sudo: bool = DEFAULT_SUDO,
    sudo_cmd: str = DEFAULT_SUDO_COMMAND,
    container_name: Optional[str] = None,
    **kwd,
) -> str:
    command_parts = _docker_prefix(
        docker_cmd=docker_cmd,
        sudo=sudo,
        sudo_cmd=sudo_cmd,
    )
    command_parts.append(command)
    command_parts.append(container_name or "{CONTAINER_NAME}")
    return " ".join(command_parts)


[docs]def build_docker_run_command(
    container_command: str,
    image: str,
    interactive: bool = False,
    terminal: bool = False,
    tag: Optional[str] = None,
    volumes: Optional[List["DockerVolume"]] = None,
    volumes_from: Optional[str] = DEFAULT_VOLUMES_FROM,
    memory: Optional[str] = DEFAULT_MEMORY,
    env_directives: Optional[List[str]] = None,
    working_directory: Optional[str] = DEFAULT_WORKING_DIRECTORY,
    name: Optional[str] = None,
    net: Optional[str] = DEFAULT_NET,
    run_extra_arguments: Optional[str] = DEFAULT_RUN_EXTRA_ARGUMENTS,
    docker_cmd: str = DEFAULT_DOCKER_COMMAND,
    sudo: bool = DEFAULT_SUDO,
    sudo_cmd: str = DEFAULT_SUDO_COMMAND,
    auto_rm: bool = DEFAULT_AUTO_REMOVE,
    set_user: Optional[str] = DEFAULT_SET_USER,
    host: Optional[str] = DEFAULT_HOST,
    guest_ports: Union[bool, str, List[str]] = False,
    container_name: Optional[str] = None,
) -> str:
    env_directives = env_directives or []
    volumes = volumes or []
    command_parts = _docker_prefix(docker_cmd=docker_cmd, sudo=sudo, sudo_cmd=sudo_cmd, host=host)
    command_parts.append("run")
    if interactive:
        command_parts.append("-i")
    if terminal:
        command_parts.append("-t")
    for env_directive in env_directives:
        # e.g. -e "GALAXY_SLOTS=$GALAXY_SLOTS"
        # These are environment variable expansions so we don't quote these.
        command_parts.extend(["-e", env_directive])
    if guest_ports is True:
        # When is True, expose all ports
        command_parts.append("-P")
    elif guest_ports:
        if not isinstance(guest_ports, list):
            guest_ports = [guest_ports]
        for guest_port in guest_ports:
            command_parts.extend(["-p", guest_port])
    if container_name:
        command_parts.extend(["--name", container_name])
    for volume in volumes:
        command_parts.extend(["-v", str(volume)])
    if volumes_from:
        command_parts.extend(["--volumes-from", shlex.quote(str(volumes_from))])
    if memory:
        command_parts.extend(["-m", shlex.quote(memory)])
    command_parts.extend(["--cpus", "${GALAXY_SLOTS:-1}"])
    if name:
        command_parts.extend(["--name", shlex.quote(name)])
    if working_directory:
        command_parts.extend(["-w", shlex.quote(working_directory)])
    if net:
        command_parts.extend(["--net", shlex.quote(net)])
    if auto_rm:
        command_parts.append("--rm")
    if run_extra_arguments:
        command_parts.append(run_extra_arguments)
    if set_user:
        user = set_user
        if set_user == DEFAULT_SET_USER:
            # If future-us is ever in here and fixing this for docker-machine just
            # use cwltool.docker_id - it takes care of this default nicely.
            euid = os.geteuid()
            egid = os.getgid()

            user = "%d:%d" % (euid, egid)
        command_parts.extend(["--user", user])
    full_image = image
    if tag:
        full_image = f"{full_image}:{tag}"
    command_parts.append(shlex.quote(full_image))
    command_parts.append(container_command)
    return " ".join(command_parts)


[docs]def command_list(command: str, command_args: Optional[List[str]] = None, **kwds) -> List[str]:
    """Return Docker command as an argv list."""
    command_args = command_args or []
    command_parts = _docker_prefix(**kwds)
    command_parts.append(command)
    command_parts.extend(command_args)
    return command_parts


[docs]def command_shell(command: str, command_args: Optional[List[str]] = None, **kwds) -> Union[str, List[str]]:
    """Return Docker command as a string for a shell or command-list."""
    command_args = command_args or []
    cmd = command_list(command, command_args, **kwds)
    to_str = kwds.get("to_str", True)
    if to_str:
        return argv_to_str(cmd)
    else:
        return cmd


def _docker_prefix(
    docker_cmd: str = DEFAULT_DOCKER_COMMAND,
    sudo: bool = DEFAULT_SUDO,
    sudo_cmd: str = DEFAULT_SUDO_COMMAND,
    host: Optional[str] = DEFAULT_HOST,
    **kwds,
) -> List[str]:
    """Prefix to issue a docker command."""
    command_parts = []
    if sudo:
        command_parts.append(sudo_cmd)
    command_parts.append(docker_cmd)
    if host:
        command_parts.extend(["-H", host])
    return command_parts


[docs]def parse_port_text(port_text):
    """

    >>> slurm_ports = parse_port_text("8888/tcp -> 0.0.0.0:32769")
    >>> slurm_ports[8888]['host']
    '0.0.0.0'
    >>> ports = parse_port_text("5432/tcp -> :::5432")
    >>> len(ports)
    0
    >>> ports_new = parse_port_text("5432/tcp -> [::]:5432")
    >>> len(ports_new)
    0
    """
    ports = None
    if port_text is not None:
        ports = {}
        for line in port_text.strip().split("\n"):
            if " -> " not in line:
                raise Exception(f"Cannot parse host and port from line [{line}]")
            tool, host = line.split(" -> ", 1)
            hostname, port = host.rsplit(":", 1)
            if hostname in ["::", "[::]"]:
                # Skip unspecified IPv6 address, which is also specified as 0:0:0:0 in another line.
                # This is brittle of course, but so is parsing the container ports like this.
                continue
            port = int(port)
            tool_p, tool_prot = tool.split("/")
            tool_p = int(tool_p)
            ports[tool_p] = dict(tool_port=tool_p, host=hostname, port=port, protocol=tool_prot)
    return ports