14182007055

Committed 31 Mar 2025 09:39PM UTC coverage: 89.45% (+3.7%) from 85.772%

Build # 14182007055

Build Type

Pull #33

github

Committed by

mocsharp

Commit Message

Add unit tests

Signed-off-by: Victor Chang <vicchang@nvidia.com>

Pull Request Pull Request #33: Enable test-app with nv-gha-runner

Run Details

46 of 51 new or added lines in 8 files covered. (90.2%)

1 existing line in 1 file now uncovered.

1789 of 2000 relevant lines covered (89.45%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.94

/src/holoscan_cli/common/dockerutils.py

# SPDX-FileCopyrightText: Copyright (c) 2023-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import logging
import os
import posixpath
import re
import subprocess
from pathlib import Path
from typing import Optional

from python_on_whales import docker

from ..common.utils import run_cmd_output
from .constants import DefaultValues, EnvironmentVariables
from .enum_types import PlatformConfiguration, SdkType
from .exceptions import GpuResourceError, InvalidManifestError, RunContainerError
from .utils import get_gpu_count, get_requested_gpus

logger = logging.getLogger("common")


def parse_docker_image_name_and_tag(
    image_name: str,
) -> tuple[Optional[str], Optional[str]]:
    """Parse a given Docker image name and tag.

    Args:
        image_name (str): Docker image name and optionally a tag

    Returns:
        Tuple[Optional[str], Optional[str]]: a tuple with first item as the name of the image
        and tag as the second item
    """
    match = re.search(
        r"^(?P<name>([\w.\-_]+((:\d+|)(?=/[a-z0-9._-]+/[a-z0-9._-]+))|)(/?)([a-z0-9.\-_/]+(/[a-z0-9.\-_]+|)))(:(?P<tag>[\w.\-_]{1,127})|)$",
        image_name,
    )

    if match is None or match.group("name") is None:
        return None, None

    name = match.group("name")
    tag = match.group("tag") if match.group("tag") else None

    return (name, tag)


def create_or_use_network(network: Optional[str], image_name: Optional[str]) -> str:
    """Create a Docker network by the given name if not already exists.

    Args:
        network (Optional[str]): name of the network to create
        image_name (Optional[str]): name of the image used to generate a network name from

    Raises:
        RunContainerError: when unable to retrieve the specified network or failed to create one.

    Returns:
        str: network name
    """
    if network is None and image_name is not None:
        network = image_name.split(":")[0]
        network += "-network"

    assert network is not None

    try:
        networks = docker.network.list(filters={"name": f"^{network}$"})
        if len(networks) > 0:
            return networks[0].name
    except Exception as ex:
        raise RunContainerError(f"error retrieving network information: {ex}") from ex

    try:
        return docker.network.create(network, driver="bridge").name
    except Exception as ex:
        raise RunContainerError(f"error creating Docker network: {ex}") from ex


def image_exists(image_name: str) -> bool:
    """Checks if the Docker image exists.

    Args:
        image_name (str): name of the Docker image

    Returns:
        bool: whether the image exists or not.
    """
    if image_name is None:
        return False
    try:
        if not docker.image.exists(image_name):
            logger.info(f"Attempting to pull image {image_name}..")
            docker.image.pull(image_name)
        return docker.image.exists(image_name)
    except Exception as e:
        logger.error(str(e))
        return False


def docker_export_tarball(file: str, tag: str):
    """Exports the docker image to a file

    Args:
        file (str): name of the exported file
        tag (str): source Docker image tag
    """
    docker.image.save(tag, file)


def create_and_get_builder(builder_name: str):
    """Creates a Docker BuildX builder

    Args:
        builder_name (str): name of the builder to create

    Returns:
        _type_: name of the builder created
    """
    builders = docker.buildx.list()
    for builder in builders:
        if builder.name == builder_name:
            logger.info(f"Using existing Docker BuildKit builder `{builder_name}`")
            return builder_name

    logger.info(
        f"Creating Docker BuildKit builder `{builder_name}` using `docker-container`"
    )
    builder = docker.buildx.create(
        name=builder_name, driver="docker-container", driver_options={"network": "host"}
    )
    return builder.name


def build_docker_image(**kwargs):
    """Builds a Docker image"""
    _ = docker.buildx.build(**kwargs)


def docker_run(
    name: str,
    image_name: str,
    input_path: Optional[Path],
    output_path: Optional[Path],
    app_info: dict,
    pkg_info: dict,
    quiet: bool,
    commands: list[str],
    health_check: bool,
    network: str,
    network_interface: Optional[str],
    use_all_nics: bool,
    gpu_enum: Optional[str],
    config: Optional[Path],
    render: bool,
    user: str,
    terminal: bool,
    devices: list[str],
    platform_config: str,
    shared_memory_size: str = "1GB",
    is_root: bool = False,
    remove: bool = False,
):
    """Creates and runs a Docker container

    `HOLOSCAN_HOSTING_SERVICE` environment variable is used for hiding the help message
    inside the tools.sh when the users run the container using holoscan run.

    Args:
        image_name (str): Docker image name
        input_path (Optional[Path]): input data path
        output_path (Optional[Path]): output data path
        app_info (dict): app manifest
        pkg_info (dict): package manifest
        quiet (bool): prints only stderr when True, otherwise, prints all logs
        commands (List[str]): list of arguments to provide to the container
        health_check (bool): whether or not to enable the gRPC health check service
        network (str): Docker network to associate the container with
        network_interface (Optional[str]): Name of the network interface for setting
            UCX_NET_DEVICES
        use_all_nics (bool): Sets UCX_CM_USE_ALL_DEVICES to 'y' if True
        config (Optional[Path]): optional configuration file for overriding the embedded one
        render (bool): whether or not to enable graphic rendering
        user (str): UID and GID to associate with the container
        terminal (bool): whether or not to enter bash terminal
        devices (List[str]): list of devices to be mapped into the container
        platformConfig (str): platform configuration value used when packaging the application,
        shared_memory_size (str): size of /dev/shm,
        is_root (bool): whether the user is root (UID = 0) or not
    """

    volumes = []
    environment_variables = {
        "NVIDIA_DRIVER_CAPABILITIES": "all",
        "HOLOSCAN_HOSTING_SERVICE": "HOLOSCAN_RUN",
        "UCX_CM_USE_ALL_DEVICES": "y" if use_all_nics else "n",
    }

    if network_interface is not None:
        environment_variables["UCX_NET_DEVICES"] = network_interface

    if health_check:
        environment_variables["HOLOSCAN_ENABLE_HEALTH_CHECK"] = "true"

    if logger.root.level == logging.DEBUG:
        environment_variables["UCX_LOG_LEVEL"] = "DEBUG"
        environment_variables["VK_LOADER_DEBUG"] = "all"

    if render:
        volumes.append(("/tmp/.X11-unix", "/tmp/.X11-unix"))
        display = os.environ.get("DISPLAY", None)
        if display is not None:
            environment_variables["DISPLAY"] = display
        xdg_session_type = os.environ.get("XDG_SESSION_TYPE", None)
        if xdg_session_type is not None:
            environment_variables["XDG_SESSION_TYPE"] = xdg_session_type
        xdg_runtime_dir = os.environ.get("XDG_RUNTIME_DIR", None)
        if xdg_runtime_dir is not None:
            volumes.append((xdg_runtime_dir, xdg_runtime_dir))
            environment_variables["XDG_RUNTIME_DIR"] = xdg_runtime_dir
        wayland_display = os.environ.get("WAYLAND_DISPLAY", None)
        if wayland_display is not None:
            environment_variables["WAYLAND_DISPLAY"] = wayland_display

    # Use user-specified --gpu values
    if gpu_enum is not None:
        environment_variables["NVIDIA_VISIBLE_DEVICES"] = gpu_enum
    # If the image was built for iGPU but the system is configured for dGPU, attempt
    # targeting the system's iGPU using the CDI spec
    elif (
        platform_config == PlatformConfiguration.iGPU.value
        and not _host_is_native_igpu()
    ):
        environment_variables["NVIDIA_VISIBLE_DEVICES"] = "nvidia.com/igpu=0"
        logger.info(
            "Attempting to run an image for iGPU (integrated GPU) on a system configured "
            "with a dGPU (discrete GPU). If this is correct (ex: IGX Orin developer kit), "
            "make sure to enable iGPU on dGPU support as described in your developer kit "
            "user guide. If not, either rebuild the image for dGPU or run this image on a "
            "system configured for iGPU only (ex: Jetson AGX, Nano...)."
        )
    # Otherwise, read specs from package manifest
    else:
        requested_gpus = get_requested_gpus(pkg_info)
        available_gpus = get_gpu_count()

        if available_gpus < requested_gpus:
            raise GpuResourceError(
                f"Available GPUs ({available_gpus}) are less than required ({requested_gpus}). "
            )

        if requested_gpus == 0:
            environment_variables["NVIDIA_VISIBLE_DEVICES"] = "all"
        else:
            environment_variables["NVIDIA_VISIBLE_DEVICES"] = ",".join(
                map(str, range(0, requested_gpus))
            )

    if "path" in app_info["input"]:
        mapped_input = Path(app_info["input"]["path"]).as_posix()
    else:
        mapped_input = DefaultValues.INPUT_DIR

    if not posixpath.isabs(mapped_input):
        mapped_input = posixpath.join(app_info["workingDirectory"], mapped_input)
    if input_path is not None:
        volumes.append((str(input_path), mapped_input))

    if "path" in app_info["output"]:
        mapped_output = Path(app_info["output"]["path"]).as_posix()
    else:
        mapped_output = DefaultValues.INPUT_DIR

    if not posixpath.isabs(mapped_output):
        mapped_output = posixpath.join(app_info["workingDirectory"], mapped_output)
    if output_path is not None:
        volumes.append((str(output_path), mapped_output))

    for env in app_info["environment"]:
        if env == EnvironmentVariables.HOLOSCAN_INPUT_PATH:
            environment_variables[env] = mapped_input
        elif env == EnvironmentVariables.HOLOSCAN_OUTPUT_PATH:
            environment_variables[env] = mapped_output
        else:
            environment_variables[env] = app_info["environment"][env]

        # always pass path to config file for Holoscan apps
        if (
            "sdk" in app_info
            and app_info["sdk"] == SdkType.Holoscan.value
            and env == EnvironmentVariables.HOLOSCAN_CONFIG_PATH
        ):
            commands.append("--config")
            commands.append(environment_variables[env])

    if config is not None:
        if EnvironmentVariables.HOLOSCAN_CONFIG_PATH not in app_info["environment"]:
            raise InvalidManifestError(
                "The application manifest does not contain a required "
                f"environment variable: '{EnvironmentVariables.HOLOSCAN_CONFIG_PATH}'"
            )
        volumes.append(
            (
                str(config),
                app_info["environment"][EnvironmentVariables.HOLOSCAN_CONFIG_PATH],
            )
        )
        logger.info(f"Using user provided configuration file: {config}")

    logger.debug(
        f"Environment variables: {json.dumps(environment_variables, indent=4, sort_keys=True)}"
    )
    logger.debug(f"Volumes: {json.dumps(volumes, indent=4, sort_keys=True)}")
    logger.debug(f"Shared memory size: {shared_memory_size}")

    ipc_mode = "host" if shared_memory_size is None else None
    ulimits = [
        "memlock=-1",
        "stack=67108864",
    ]
    additional_devices, group_adds = _additional_devices_to_mount(is_root)
    devices.extend(additional_devices)

    video_group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "video").split(":")[2]
    if not is_root and video_group not in group_adds:
        group_adds.append(video_group)

    if terminal:
        _enter_terminal(
            name,
            image_name,
            app_info,
            network,
            user,
            volumes,
            environment_variables,
            shared_memory_size,
            ipc_mode,
            ulimits,
            devices,
            group_adds,
            remove,
        )
    else:
        _start_container(
            name,
            image_name,
            app_info,
            quiet,
            commands,
            network,
            user,
            volumes,
            environment_variables,
            shared_memory_size,
            ipc_mode,
            ulimits,
            devices,
            group_adds,
            remove,
        )


def _start_container(
    name,
    image_name,
    app_info,
    quiet,
    commands,
    network,
    user,
    volumes,
    environment_variables,
    shared_memory_size,
    ipc_mode,
    ulimits,
    devices,
    group_adds,
    remove,
):
    container = docker.container.create(
        image_name,
        command=commands,
        envs=environment_variables,
        hostname=name,
        name=name,
        networks=[network],
        remove=False,
        shm_size=shared_memory_size,
        user=user,
        volumes=volumes,
        workdir=app_info["workingDirectory"],
        ipc=ipc_mode,
        cap_add=["CAP_SYS_PTRACE"],
        ulimit=ulimits,
        devices=devices,
        groups_add=group_adds,
        runtime="nvidia",
    )
    container_name = container.name
    container_id = container.id[:12]

    ulimit_str = ", ".join(
        f"{ulimit.name}={ulimit.soft}:{ulimit.hard}"
        for ulimit in container.host_config.ulimits
    )
    logger.info(
        f"Launching container ({container_id}) using image '{image_name}'..."
        f"\n    container name:      {container_name}"
        f"\n    host name:           {container.config.hostname}"
        f"\n    network:             {network}"
        f"\n    user:                {container.config.user}"
        f"\n    ulimits:             {ulimit_str}"
        f"\n    cap_add:             {', '.join(container.host_config.cap_add)}"
        f"\n    ipc mode:            {container.host_config.ipc_mode}"
        f"\n    shared memory size:  {container.host_config.shm_size}"
        f"\n    devices:             {', '.join(devices)}"
        f"\n    group_add:           {', '.join(group_adds)}"
    )
    logs = container.start(
        attach=True,
        stream=True,
    )

    for log in logs:
        if log[0] == "stdout":
            if not quiet:
                print(log[1].decode("utf-8"))
        elif log[0] == "stderr":
            try:
                print(str(log[1].decode("utf-8")))
            except Exception:
                print(str(log[1]))

    exit_code = container.state.exit_code
    logger.info(
        f"Container '{container_name}'({container_id}) exited with code {exit_code}."
    )

    if remove:
        container.remove()

    if exit_code != 0:
        raise RuntimeError(
            f"Container '{container_name}'({container_id}) exited with code {exit_code}."
        )


def _enter_terminal(
    name,
    image_name,
    app_info,
    network,
    user,
    volumes,
    environment_variables,
    shared_memory_size,
    ipc_mode,
    ulimits,
    devices,
    group_adds,
    remove,
):
    print("\n\nEntering terminal...")
    print(
        "\n".join(
            f"\t{k:25s}\t{v}"
            for k, v in sorted(environment_variables.items(), key=lambda t: str(t[0]))
        )
    )
    print("\n\n")
    docker.container.run(
        image_name,
        detach=False,
        entrypoint="/bin/bash",
        envs=environment_variables,
        hostname=name,
        interactive=True,
        name=name,
        networks=[network],
        remove=remove,
        shm_size=shared_memory_size,
        tty=True,
        user=user,
        volumes=volumes,
        workdir=app_info["workingDirectory"],
        ipc=ipc_mode,
        cap_add=["CAP_SYS_PTRACE"],
        ulimit=ulimits,
        devices=devices,
        groups_add=group_adds,
        runtime="nvidia",
    )
    logger.info("Container exited.")


def _additional_devices_to_mount(is_root: bool):
    """Mounts additional devices"""
    devices = []
    group_adds = []

    # On iGPU, the /dev/dri/* devices (mounted by the NV container runtime) permissions require root
    # privilege or to be part of the `video` and `render` groups. The ID for these group names might
    # differ on the host system and in the container, so we need to pass the group ID instead of the
    # group name when running docker.
    if (
        os.path.exists("/sys/devices/platform/gpu.0/load")
        and os.path.exists("/usr/bin/tegrastats")
        and not is_root
    ):
        group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "video").split(":")[2]
        group_adds.append(group)
        group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "render").split(":")[2]
        group_adds.append(group)
    return (devices, group_adds)


def _host_is_native_igpu() -> bool:
    proc = subprocess.run(
        ["nvidia-smi", "--query-gpu", "name", "--format=csv,noheader"],
        shell=False,
        capture_output=True,
    )
    return "nvgpu" in str(proc.stdout)

1	# SPDX-FileCopyrightText: Copyright (c) 2023-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2	# SPDX-License-Identifier: Apache-2.0
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	import json	1✔
16	import logging	1✔
17	import os	1✔
18	import posixpath	1✔
19	import re	1✔
20	import subprocess	1✔
21	from pathlib import Path	1✔
22	from typing import Optional	1✔
23
24	from python_on_whales import docker	1✔
25
26	from ..common.utils import run_cmd_output	1✔
27	from .constants import DefaultValues, EnvironmentVariables	1✔
28	from .enum_types import PlatformConfiguration, SdkType	1✔
29	from .exceptions import GpuResourceError, InvalidManifestError, RunContainerError	1✔
30	from .utils import get_gpu_count, get_requested_gpus	1✔
31
32	logger = logging.getLogger("common")	1✔
33
34
35	def parse_docker_image_name_and_tag(	1✔
36	image_name: str,
37	) -> tuple[Optional[str], Optional[str]]:
38	"""Parse a given Docker image name and tag.
39
40	Args:
41	image_name (str): Docker image name and optionally a tag
42
43	Returns:
44	Tuple[Optional[str], Optional[str]]: a tuple with first item as the name of the image
45	and tag as the second item
46	"""
47	match = re.search(	1✔
48	r"^(?P<name>([\w.\-_]+((:\d+\|)(?=/[a-z0-9._-]+/[a-z0-9._-]+))\|)(/?)([a-z0-9.\-_/]+(/[a-z0-9.\-_]+\|)))(:(?P<tag>[\w.\-_]{1,127})\|)$",
49	image_name,
50	)
51
52	if match is None or match.group("name") is None:	1✔
53	return None, None	1✔
54
55	name = match.group("name")	1✔
56	tag = match.group("tag") if match.group("tag") else None	1✔
57
58	return (name, tag)	1✔
59
60
61	def create_or_use_network(network: Optional[str], image_name: Optional[str]) -> str:	1✔
62	"""Create a Docker network by the given name if not already exists.
63
64	Args:
65	network (Optional[str]): name of the network to create
66	image_name (Optional[str]): name of the image used to generate a network name from
67
68	Raises:
69	RunContainerError: when unable to retrieve the specified network or failed to create one.
70
71	Returns:
72	str: network name
73	"""
74	if network is None and image_name is not None:	1✔
75	network = image_name.split(":")[0]	1✔
76	network += "-network"	1✔
77
78	assert network is not None	1✔
79
80	try:	1✔
81	networks = docker.network.list(filters={"name": f"^{network}$"})	1✔
82	if len(networks) > 0:	1✔
83	return networks[0].name	1✔
84	except Exception as ex:	1✔
85	raise RunContainerError(f"error retrieving network information: {ex}") from ex	1✔
86
87	try:	1✔
88	return docker.network.create(network, driver="bridge").name	1✔
89	except Exception as ex:	×
90	raise RunContainerError(f"error creating Docker network: {ex}") from ex	×
91
92
93	def image_exists(image_name: str) -> bool:	1✔
94	"""Checks if the Docker image exists.
95
96	Args:
97	image_name (str): name of the Docker image
98
99	Returns:
100	bool: whether the image exists or not.
101	"""
102	if image_name is None:	1✔
103	return False	1✔
104	try:	1✔
105	if not docker.image.exists(image_name):	1✔
106	logger.info(f"Attempting to pull image {image_name}..")	1✔
107	docker.image.pull(image_name)	1✔
108	return docker.image.exists(image_name)	1✔
109	except Exception as e:	1✔
110	logger.error(str(e))	1✔
111	return False	1✔
112
113
114	def docker_export_tarball(file: str, tag: str):	1✔
115	"""Exports the docker image to a file
116
117	Args:
118	file (str): name of the exported file
119	tag (str): source Docker image tag
120	"""
121	docker.image.save(tag, file)	1✔
122
123
124	def create_and_get_builder(builder_name: str):	1✔
125	"""Creates a Docker BuildX builder
126
127	Args:
128	builder_name (str): name of the builder to create
129
130	Returns:
131	_type_: name of the builder created
132	"""
133	builders = docker.buildx.list()	1✔
134	for builder in builders:	1✔
135	if builder.name == builder_name:	1✔
136	logger.info(f"Using existing Docker BuildKit builder `{builder_name}`")	1✔
137	return builder_name	1✔
138
139	logger.info(	1✔
140	f"Creating Docker BuildKit builder `{builder_name}` using `docker-container`"
141	)
142	builder = docker.buildx.create(	1✔
143	name=builder_name, driver="docker-container", driver_options={"network": "host"}
144	)
145	return builder.name	1✔
146
147
148	def build_docker_image(**kwargs):	1✔
149	"""Builds a Docker image"""
150	_ = docker.buildx.build(**kwargs)	×
151
152
153	def docker_run(	1✔
154	name: str,
155	image_name: str,
156	input_path: Optional[Path],
157	output_path: Optional[Path],
158	app_info: dict,
159	pkg_info: dict,
160	quiet: bool,
161	commands: list[str],
162	health_check: bool,
163	network: str,
164	network_interface: Optional[str],
165	use_all_nics: bool,
166	gpu_enum: Optional[str],
167	config: Optional[Path],
168	render: bool,
169	user: str,
170	terminal: bool,
171	devices: list[str],
172	platform_config: str,
173	shared_memory_size: str = "1GB",
174	is_root: bool = False,
175	remove: bool = False,
176	):
177	"""Creates and runs a Docker container
178
179	`HOLOSCAN_HOSTING_SERVICE` environment variable is used for hiding the help message
180	inside the tools.sh when the users run the container using holoscan run.
181
182	Args:
183	image_name (str): Docker image name
184	input_path (Optional[Path]): input data path
185	output_path (Optional[Path]): output data path
186	app_info (dict): app manifest
187	pkg_info (dict): package manifest
188	quiet (bool): prints only stderr when True, otherwise, prints all logs
189	commands (List[str]): list of arguments to provide to the container
190	health_check (bool): whether or not to enable the gRPC health check service
191	network (str): Docker network to associate the container with
192	network_interface (Optional[str]): Name of the network interface for setting
193	UCX_NET_DEVICES
194	use_all_nics (bool): Sets UCX_CM_USE_ALL_DEVICES to 'y' if True
195	config (Optional[Path]): optional configuration file for overriding the embedded one
196	render (bool): whether or not to enable graphic rendering
197	user (str): UID and GID to associate with the container
198	terminal (bool): whether or not to enter bash terminal
199	devices (List[str]): list of devices to be mapped into the container
200	platformConfig (str): platform configuration value used when packaging the application,
201	shared_memory_size (str): size of /dev/shm,
202	is_root (bool): whether the user is root (UID = 0) or not
203	"""
204
205	volumes = []	1✔
206	environment_variables = {	1✔
207	"NVIDIA_DRIVER_CAPABILITIES": "all",
208	"HOLOSCAN_HOSTING_SERVICE": "HOLOSCAN_RUN",
209	"UCX_CM_USE_ALL_DEVICES": "y" if use_all_nics else "n",
210	}
211
212	if network_interface is not None:	1✔
213	environment_variables["UCX_NET_DEVICES"] = network_interface	×
214
215	if health_check:	1✔
216	environment_variables["HOLOSCAN_ENABLE_HEALTH_CHECK"] = "true"	×
217
218	if logger.root.level == logging.DEBUG:	1✔
219	environment_variables["UCX_LOG_LEVEL"] = "DEBUG"	×
220	environment_variables["VK_LOADER_DEBUG"] = "all"	×
221
222	if render:	1✔
223	volumes.append(("/tmp/.X11-unix", "/tmp/.X11-unix"))	1✔
224	display = os.environ.get("DISPLAY", None)	1✔
225	if display is not None:	1✔
226	environment_variables["DISPLAY"] = display	1✔
227	xdg_session_type = os.environ.get("XDG_SESSION_TYPE", None)	1✔
228	if xdg_session_type is not None:	1✔
229	environment_variables["XDG_SESSION_TYPE"] = xdg_session_type	1✔
230	xdg_runtime_dir = os.environ.get("XDG_RUNTIME_DIR", None)	1✔
231	if xdg_runtime_dir is not None:	1✔
232	volumes.append((xdg_runtime_dir, xdg_runtime_dir))	1✔
233	environment_variables["XDG_RUNTIME_DIR"] = xdg_runtime_dir	1✔
234	wayland_display = os.environ.get("WAYLAND_DISPLAY", None)	1✔
235	if wayland_display is not None:	1✔
236	environment_variables["WAYLAND_DISPLAY"] = wayland_display	1✔
237
238	# Use user-specified --gpu values
239	if gpu_enum is not None:	1✔
240	environment_variables["NVIDIA_VISIBLE_DEVICES"] = gpu_enum	×
241	# If the image was built for iGPU but the system is configured for dGPU, attempt
242	# targeting the system's iGPU using the CDI spec
243	elif (	1✔
244	platform_config == PlatformConfiguration.iGPU.value
245	and not _host_is_native_igpu()
246	):
247	environment_variables["NVIDIA_VISIBLE_DEVICES"] = "nvidia.com/igpu=0"	1✔
248	logger.info(	1✔
249	"Attempting to run an image for iGPU (integrated GPU) on a system configured "
250	"with a dGPU (discrete GPU). If this is correct (ex: IGX Orin developer kit), "
251	"make sure to enable iGPU on dGPU support as described in your developer kit "
252	"user guide. If not, either rebuild the image for dGPU or run this image on a "
253	"system configured for iGPU only (ex: Jetson AGX, Nano...)."
254	)
255	# Otherwise, read specs from package manifest
256	else:
257	requested_gpus = get_requested_gpus(pkg_info)	1✔
258	available_gpus = get_gpu_count()	1✔
259
260	if available_gpus < requested_gpus:	1✔
261	raise GpuResourceError(	1✔
262	f"Available GPUs ({available_gpus}) are less than required ({requested_gpus}). "
263	)
264
265	if requested_gpus == 0:	1✔
266	environment_variables["NVIDIA_VISIBLE_DEVICES"] = "all"	×
267	else:
268	environment_variables["NVIDIA_VISIBLE_DEVICES"] = ",".join(	1✔
269	map(str, range(0, requested_gpus))
270	)
271
272	if "path" in app_info["input"]:	1✔
273	mapped_input = Path(app_info["input"]["path"]).as_posix()	1✔
274	else:
275	mapped_input = DefaultValues.INPUT_DIR	×
276
277	if not posixpath.isabs(mapped_input):	1✔
278	mapped_input = posixpath.join(app_info["workingDirectory"], mapped_input)	×
279	if input_path is not None:	1✔
280	volumes.append((str(input_path), mapped_input))	1✔
281
282	if "path" in app_info["output"]:	1✔
283	mapped_output = Path(app_info["output"]["path"]).as_posix()	1✔
284	else:
285	mapped_output = DefaultValues.INPUT_DIR	×
286
287	if not posixpath.isabs(mapped_output):	1✔
288	mapped_output = posixpath.join(app_info["workingDirectory"], mapped_output)	×
289	if output_path is not None:	1✔
290	volumes.append((str(output_path), mapped_output))	1✔
291
292	for env in app_info["environment"]:	1✔
293	if env == EnvironmentVariables.HOLOSCAN_INPUT_PATH:	1✔
294	environment_variables[env] = mapped_input	1✔
295	elif env == EnvironmentVariables.HOLOSCAN_OUTPUT_PATH:	1✔
296	environment_variables[env] = mapped_output	1✔
297	else:
298	environment_variables[env] = app_info["environment"][env]	1✔
299
300	# always pass path to config file for Holoscan apps
301	if (	1✔
302	"sdk" in app_info
303	and app_info["sdk"] == SdkType.Holoscan.value
304	and env == EnvironmentVariables.HOLOSCAN_CONFIG_PATH
305	):
306	commands.append("--config")	1✔
307	commands.append(environment_variables[env])	1✔
308
309	if config is not None:	1✔
310	if EnvironmentVariables.HOLOSCAN_CONFIG_PATH not in app_info["environment"]:	1✔
311	raise InvalidManifestError(	×
312	"The application manifest does not contain a required "
313	f"environment variable: '{EnvironmentVariables.HOLOSCAN_CONFIG_PATH}'"
314	)
315	volumes.append(	1✔
316	(
317	str(config),
318	app_info["environment"][EnvironmentVariables.HOLOSCAN_CONFIG_PATH],
319	)
320	)
321	logger.info(f"Using user provided configuration file: {config}")	1✔
322
323	logger.debug(	1✔
324	f"Environment variables: {json.dumps(environment_variables, indent=4, sort_keys=True)}"
325	)
326	logger.debug(f"Volumes: {json.dumps(volumes, indent=4, sort_keys=True)}")	1✔
327	logger.debug(f"Shared memory size: {shared_memory_size}")	1✔
328
329	ipc_mode = "host" if shared_memory_size is None else None	1✔
330	ulimits = [	1✔
331	"memlock=-1",
332	"stack=67108864",
333	]
334	additional_devices, group_adds = _additional_devices_to_mount(is_root)	1✔
335	devices.extend(additional_devices)	1✔
336
337	video_group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "video").split(":")[2]	1✔
338	if not is_root and video_group not in group_adds:	1✔
339	group_adds.append(video_group)	1✔
340
341	if terminal:	1✔
342	_enter_terminal(	1✔
343	name,
344	image_name,
345	app_info,
346	network,
347	user,
348	volumes,
349	environment_variables,
350	shared_memory_size,
351	ipc_mode,
352	ulimits,
353	devices,
354	group_adds,
355	remove,
356	)
357	else:
358	_start_container(	1✔
359	name,
360	image_name,
361	app_info,
362	quiet,
363	commands,
364	network,
365	user,
366	volumes,
367	environment_variables,
368	shared_memory_size,
369	ipc_mode,
370	ulimits,
371	devices,
372	group_adds,
373	remove,
374	)
375
376
377	def _start_container(	1✔
378	name,
379	image_name,
380	app_info,
381	quiet,
382	commands,
383	network,
384	user,
385	volumes,
386	environment_variables,
387	shared_memory_size,
388	ipc_mode,
389	ulimits,
390	devices,
391	group_adds,
392	remove,
393	):
394	container = docker.container.create(	1✔
395	image_name,
396	command=commands,
397	envs=environment_variables,
398	hostname=name,
399	name=name,
400	networks=[network],
401	remove=False,
402	shm_size=shared_memory_size,
403	user=user,
404	volumes=volumes,
405	workdir=app_info["workingDirectory"],
406	ipc=ipc_mode,
407	cap_add=["CAP_SYS_PTRACE"],
408	ulimit=ulimits,
409	devices=devices,
410	groups_add=group_adds,
411	runtime="nvidia",
412	)
413	container_name = container.name	1✔
414	container_id = container.id[:12]	1✔
415
416	ulimit_str = ", ".join(	1✔
417	f"{ulimit.name}={ulimit.soft}:{ulimit.hard}"
418	for ulimit in container.host_config.ulimits
419	)
420	logger.info(	1✔
421	f"Launching container ({container_id}) using image '{image_name}'..."
422	f"\n container name: {container_name}"
423	f"\n host name: {container.config.hostname}"
424	f"\n network: {network}"
425	f"\n user: {container.config.user}"
426	f"\n ulimits: {ulimit_str}"
427	f"\n cap_add: {', '.join(container.host_config.cap_add)}"
428	f"\n ipc mode: {container.host_config.ipc_mode}"
429	f"\n shared memory size: {container.host_config.shm_size}"
430	f"\n devices: {', '.join(devices)}"
431	f"\n group_add: {', '.join(group_adds)}"
432	)
433	logs = container.start(	1✔
434	attach=True,
435	stream=True,
436	)
437
438	for log in logs:	1✔
439	if log[0] == "stdout":	1✔
440	if not quiet:	1✔
441	print(log[1].decode("utf-8"))	1✔
442	elif log[0] == "stderr":	1✔
443	try:	1✔
444	print(str(log[1].decode("utf-8")))	1✔
445	except Exception:	×
446	print(str(log[1]))	×
447
448	exit_code = container.state.exit_code	1✔
449	logger.info(	1✔
450	f"Container '{container_name}'({container_id}) exited with code {exit_code}."
451	)
452
453	if remove:	1✔
NEW 454	container.remove()	×
455
456	if exit_code != 0:	1✔
NEW 457	raise RuntimeError(	×
458	f"Container '{container_name}'({container_id}) exited with code {exit_code}."
459	)
460
461
462	def _enter_terminal(	1✔
463	name,
464	image_name,
465	app_info,
466	network,
467	user,
468	volumes,
469	environment_variables,
470	shared_memory_size,
471	ipc_mode,
472	ulimits,
473	devices,
474	group_adds,
475	remove,
476	):
477	print("\n\nEntering terminal...")	1✔
478	print(	1✔
479	"\n".join(
480	f"\t{k:25s}\t{v}"
481	for k, v in sorted(environment_variables.items(), key=lambda t: str(t[0]))
482	)
483	)
484	print("\n\n")	1✔
485	docker.container.run(	1✔
486	image_name,
487	detach=False,
488	entrypoint="/bin/bash",
489	envs=environment_variables,
490	hostname=name,
491	interactive=True,
492	name=name,
493	networks=[network],
494	remove=remove,
495	shm_size=shared_memory_size,
496	tty=True,
497	user=user,
498	volumes=volumes,
499	workdir=app_info["workingDirectory"],
500	ipc=ipc_mode,
501	cap_add=["CAP_SYS_PTRACE"],
502	ulimit=ulimits,
503	devices=devices,
504	groups_add=group_adds,
505	runtime="nvidia",
506	)
507	logger.info("Container exited.")	1✔
508
509
510	def _additional_devices_to_mount(is_root: bool):	1✔
511	"""Mounts additional devices"""
512	devices = []	1✔
513	group_adds = []	1✔
514
515	# On iGPU, the /dev/dri/* devices (mounted by the NV container runtime) permissions require root
516	# privilege or to be part of the `video` and `render` groups. The ID for these group names might
517	# differ on the host system and in the container, so we need to pass the group ID instead of the
518	# group name when running docker.
519	if (	1✔
520	os.path.exists("/sys/devices/platform/gpu.0/load")
521	and os.path.exists("/usr/bin/tegrastats")
522	and not is_root
523	):
524	group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "video").split(":")[2]	1✔
525	group_adds.append(group)	1✔
526	group = run_cmd_output(["/usr/bin/cat", "/etc/group"], "render").split(":")[2]	1✔
527	group_adds.append(group)	1✔
528	return (devices, group_adds)	1✔
529
530
531	def _host_is_native_igpu() -> bool:	1✔
532	proc = subprocess.run(	1✔
533	["nvidia-smi", "--query-gpu", "name", "--format=csv,noheader"],
534	shell=False,
535	capture_output=True,
536	)
537	return "nvgpu" in str(proc.stdout)	1✔

nvidia-holoscan / holoscan-cli / 14182007055

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous