14097962944

Committed 27 Mar 2025 03:06AM UTC coverage: 85.7% (-0.07%) from 85.772%

Build # 14097962944

Build Type

Pull #33

github

Committed by

mocsharp

Commit Message

f

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

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

Run Details

35 of 43 new or added lines in 6 files covered. (81.4%)

2 existing lines in 2 files now uncovered.

1708 of 1993 relevant lines covered (85.7%)

0.86 hits per line

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

91.03

/src/holoscan_cli/runner/runner.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 re
import shutil
import sys
import tempfile
from argparse import Namespace
from glob import glob
from pathlib import Path
from typing import Optional

from ..common.dockerutils import create_or_use_network, docker_run, image_exists
from ..common.exceptions import ManifestReadError, UnmatchedDeviceError
from ..common.utils import (
    compare_versions,
    get_requested_gpus,
    print_manifest_json,
    run_cmd,
    run_cmd_output,
)
from .resources import get_shared_memory_size

logger = logging.getLogger("runner")


def _fetch_map_manifest(map_name: str) -> tuple[dict, dict]:
    """
    Execute HAP/MAP and fetch the manifest files.

    Args:
        map_name: HAP/MAP image name.

    Returns:
        app_info: application manifest as a python dict
        pkg_info: package manifest as a python dict
    """

    def _ensure_valid_exit_code(returncode: int) -> None:
        if returncode != 0:
            raise ManifestReadError("Error reading manifest file from the package.")

    logger.info("Reading HAP/MAP manifest...")

    with tempfile.TemporaryDirectory() as info_dir:
        docker_id = run_cmd_output(["docker", "create", map_name]).strip()
        returncode = run_cmd(
            [
                "docker",
                "cp",
                f"{docker_id}:/etc/holoscan/app.json",
                f"{info_dir}/app.json",
            ]
        )
        _ensure_valid_exit_code(returncode)
        returncode = run_cmd(
            [
                "docker",
                "cp",
                f"{docker_id}:/etc/holoscan/pkg.json",
                f"{info_dir}/pkg.json",
            ]
        )
        _ensure_valid_exit_code(returncode)
        returncode = run_cmd(["docker", "rm", "-v", docker_id])
        _ensure_valid_exit_code(returncode)

        app_json = Path(f"{info_dir}/app.json")
        pkg_json = Path(f"{info_dir}/pkg.json")

        app_info = json.loads(app_json.read_text())
        pkg_info = json.loads(pkg_json.read_text())

        print_manifest_json(app_info, "app.json")
        print_manifest_json(pkg_info, "pkg.json")

        return app_info, pkg_info


def _run_app(args: Namespace, app_info: dict, pkg_info: dict):
    """
    Executes the Holoscan Application Package.

    Args:
        args: user arguments
        app_info: application manifest dictionary
        pkg_info: package manifest dictionary

    Returns:
        returncode: command returncode
    """

    map_name: str = args.map
    input_path: Path = args.input
    output_path: Path = args.output
    quiet: bool = args.quiet
    driver: bool = args.driver
    worker: bool = args.worker
    health_check: bool = args.health_check
    fragments: Optional[str] = args.fragments
    network: str = create_or_use_network(args.network, map_name)
    nic: Optional[str] = args.nic if args.nic else None
    use_all_nics: bool = args.use_all_nics
    gpus: Optional[str] = args.gpus if args.gpus else None
    config: Optional[Path] = args.config if args.config else None
    address: Optional[str] = args.address if args.address else None
    worker_address: Optional[str] = args.worker_address if args.worker_address else None
    render: bool = args.render
    user: str = f"{args.uid}:{args.gid}"
    hostname: Optional[str] = "driver" if driver else None
    terminal: bool = args.terminal
    platform_config: str = pkg_info.get("platformConfig")
    shared_memory_size: Optional[str] = (
        get_shared_memory_size(pkg_info, worker, driver, fragments, args.shm_size)
        if args.shm_size
        else None
    )

    commands = []
    devices = _lookup_devices(args.device) if args.device is not None else []

    if driver:
        commands.append("--driver")
        logger.info("Application running in Driver mode")
    if worker:
        commands.append("--worker")
        logger.info("Application running in Worker mode")
    if fragments:
        commands.append("--fragments")
        commands.append(fragments)
        logger.info(f"Configured fragments: {fragments}")
    if address:
        commands.append("--address")
        commands.append(address)
        logger.info(f"App Driver address and port: {address}")
    if worker_address:
        commands.append("--worker_address")
        commands.append(worker_address)
        logger.info(f"App Worker address and port: {worker_address}")

    docker_run(
        hostname,
        map_name,
        input_path,
        output_path,
        app_info,
        pkg_info,
        quiet,
        commands,
        health_check,
        network,
        nic,
        use_all_nics,
        gpus,
        config,
        render,
        user,
        terminal,
        devices,
        platform_config,
        shared_memory_size,
        args.uid == 0,
    )


def _lookup_devices(devices: list[str]) -> list[str]:
    """
    Looks up matching devices in /dev and returns a list
    of fully qualified device paths.

    Raises exception if any devices cannot be found.
    """
    matched_devices = []
    unmatched_devices = []

    for dev in devices:
        pattern = dev
        if not pattern.startswith("/"):
            pattern = "/dev/" + pattern
        found_devices = glob(pattern)
        if len(found_devices) == 0:
            unmatched_devices.append(dev)
        else:
            matched_devices.extend(found_devices)

    if len(unmatched_devices) > 0:
        raise UnmatchedDeviceError(unmatched_devices)

    return matched_devices


def _dependency_verification(map_name: str) -> bool:
    """Check if all the dependencies are installed or not.

    Args:
        map_name: HAP/MAP name

    Returns:
        True if all dependencies are satisfied, otherwise False.
    """
    logger.info("Checking dependencies...")

    # check for docker
    prog = "docker"
    logger.info(f'--> Verifying if "{prog}" is installed...\n')
    if not shutil.which(prog):
        logger.error(
            '"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
            prog,
        )
        return False

    buildx_paths = [
        os.path.expandvars("$HOME/.docker/cli-plugins"),
        "/usr/local/lib/docker/cli-plugins",
        "/usr/local/libexec/docker/cli-plugins",
        "/usr/lib/docker/cli-plugins",
        "/usr/libexec/docker/cli-plugins",
    ]
    prog = "docker-buildx"
    logger.info('--> Verifying if "%s" is installed...\n', prog)
    buildx_found = False
    for path in buildx_paths:
        if shutil.which(os.path.join(path, prog)):
            buildx_found = True

    if not buildx_found:
        logger.error(
            '"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
            prog,
        )
        return False

    # check for map image
    logger.info('--> Verifying if "%s" is available...\n', map_name)
    if not image_exists(map_name):
        logger.error("Unable to fetch required image.")
        return False

    return True


def _pkg_specific_dependency_verification(pkg_info: dict) -> bool:
    """Checks for any package specific dependencies.

    Currently it verifies the following dependencies:
    * If gpu has been requested by the application, verify that nvidia-ctk is installed.
    Note: when running inside a Docker container, always assume nvidia-ctk is installed.
    Args:
        pkg_info: package manifest as a python dict

    Returns:
        True if all dependencies are satisfied, otherwise False.
    """
    if os.path.exists("/.dockerenv"):
        logger.info("--> Skipping nvidia-ctk check inside Docker...\n")
        return True

    requested_gpus = get_requested_gpus(pkg_info)
    if requested_gpus > 0:
        # check for NVIDIA Container TOolkit
        prog = "nvidia-ctk"
        logger.info('--> Verifying if "%s" is installed...\n', prog)
        if not shutil.which(prog):
            logger.error(
                '"%s" not installed, please install NVIDIA Container Toolkit.', prog
            )
            return False

        logger.info('--> Verifying "%s" version...\n', prog)
        output = run_cmd_output(["nvidia-ctk", "--version"], "version")
        match = re.search(r"([0-9]+\.[0-9]+\.[0-9]+)", output)
        min_ctk_version = "1.12.0"
        recommended_ctk_version = "1.14.1"

        if match is None:
            logger.error(
                f"Error detecting NVIDIA Container Toolkit version. "
                f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
            )
            return False
        elif compare_versions(min_ctk_version, match.group()) > 0:
            logger.error(
                f"Found '{prog}' Version {match.group()}. "
                f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
            )
            return False

    return True


def execute_run_command(args: Namespace):
    """
    Entrypoint for the Holoscan Run command.

    Args:
        args: Namespace object containing user arguments.

    Returns:
        None
    """
    if not _dependency_verification(args.map):
        logger.error("Execution Aborted")
        sys.exit(2)
    try:
        # Fetch application manifest from MAP
        app_info, pkg_info = _fetch_map_manifest(args.map)
    except Exception as e:
        logger.error(f"Failed to fetch MAP manifest: {e}")
        sys.exit(2)

    if not _pkg_specific_dependency_verification(pkg_info):
        logger.error("Execution Aborted")
        sys.exit(2)
    try:
        # Run Holoscan Application
        _run_app(args, app_info, pkg_info)
    except Exception as ex:
        logger.debug(ex, exc_info=True)
        logger.error(f"Error executing {args.map}: {ex}")
        sys.exit(2)

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 re	1✔
19	import shutil	1✔
20	import sys	1✔
21	import tempfile	1✔
22	from argparse import Namespace	1✔
23	from glob import glob	1✔
24	from pathlib import Path	1✔
25	from typing import Optional	1✔
26
27	from ..common.dockerutils import create_or_use_network, docker_run, image_exists	1✔
28	from ..common.exceptions import ManifestReadError, UnmatchedDeviceError	1✔
29	from ..common.utils import (	1✔
30	compare_versions,
31	get_requested_gpus,
32	print_manifest_json,
33	run_cmd,
34	run_cmd_output,
35	)
36	from .resources import get_shared_memory_size	1✔
37
38	logger = logging.getLogger("runner")	1✔
39
40
41	def _fetch_map_manifest(map_name: str) -> tuple[dict, dict]:	1✔
42	"""
43	Execute HAP/MAP and fetch the manifest files.
44
45	Args:
46	map_name: HAP/MAP image name.
47
48	Returns:
49	app_info: application manifest as a python dict
50	pkg_info: package manifest as a python dict
51	"""
52
53	def _ensure_valid_exit_code(returncode: int) -> None:	1✔
54	if returncode != 0:	1✔
55	raise ManifestReadError("Error reading manifest file from the package.")	1✔
56
57	logger.info("Reading HAP/MAP manifest...")	1✔
58
59	with tempfile.TemporaryDirectory() as info_dir:	1✔
60	docker_id = run_cmd_output(["docker", "create", map_name]).strip()	1✔
61	returncode = run_cmd(	1✔
62	[
63	"docker",
64	"cp",
65	f"{docker_id}:/etc/holoscan/app.json",
66	f"{info_dir}/app.json",
67	]
68	)
69	_ensure_valid_exit_code(returncode)	1✔
70	returncode = run_cmd(	1✔
71	[
72	"docker",
73	"cp",
74	f"{docker_id}:/etc/holoscan/pkg.json",
75	f"{info_dir}/pkg.json",
76	]
77	)
78	_ensure_valid_exit_code(returncode)	1✔
79	returncode = run_cmd(["docker", "rm", "-v", docker_id])	1✔
80	_ensure_valid_exit_code(returncode)	1✔
81
82	app_json = Path(f"{info_dir}/app.json")	1✔
83	pkg_json = Path(f"{info_dir}/pkg.json")	1✔
84
85	app_info = json.loads(app_json.read_text())	1✔
86	pkg_info = json.loads(pkg_json.read_text())	1✔
87
88	print_manifest_json(app_info, "app.json")	1✔
89	print_manifest_json(pkg_info, "pkg.json")	1✔
90
91	return app_info, pkg_info	1✔
92
93
94	def _run_app(args: Namespace, app_info: dict, pkg_info: dict):	1✔
95	"""
96	Executes the Holoscan Application Package.
97
98	Args:
99	args: user arguments
100	app_info: application manifest dictionary
101	pkg_info: package manifest dictionary
102
103	Returns:
104	returncode: command returncode
105	"""
106
107	map_name: str = args.map	1✔
108	input_path: Path = args.input	1✔
109	output_path: Path = args.output	1✔
110	quiet: bool = args.quiet	1✔
111	driver: bool = args.driver	1✔
112	worker: bool = args.worker	1✔
113	health_check: bool = args.health_check	1✔
114	fragments: Optional[str] = args.fragments	1✔
115	network: str = create_or_use_network(args.network, map_name)	1✔
116	nic: Optional[str] = args.nic if args.nic else None	1✔
117	use_all_nics: bool = args.use_all_nics	1✔
118	gpus: Optional[str] = args.gpus if args.gpus else None	1✔
119	config: Optional[Path] = args.config if args.config else None	1✔
120	address: Optional[str] = args.address if args.address else None	1✔
121	worker_address: Optional[str] = args.worker_address if args.worker_address else None	1✔
122	render: bool = args.render	1✔
123	user: str = f"{args.uid}:{args.gid}"	1✔
124	hostname: Optional[str] = "driver" if driver else None	1✔
125	terminal: bool = args.terminal	1✔
126	platform_config: str = pkg_info.get("platformConfig")	1✔
127	shared_memory_size: Optional[str] = (	1✔
128	get_shared_memory_size(pkg_info, worker, driver, fragments, args.shm_size)
129	if args.shm_size
130	else None
131	)
132
133	commands = []	1✔
134	devices = _lookup_devices(args.device) if args.device is not None else []	1✔
135
136	if driver:	1✔
137	commands.append("--driver")	1✔
138	logger.info("Application running in Driver mode")	1✔
139	if worker:	1✔
140	commands.append("--worker")	1✔
141	logger.info("Application running in Worker mode")	1✔
142	if fragments:	1✔
143	commands.append("--fragments")	1✔
144	commands.append(fragments)	1✔
145	logger.info(f"Configured fragments: {fragments}")	1✔
146	if address:	1✔
147	commands.append("--address")	1✔
148	commands.append(address)	1✔
149	logger.info(f"App Driver address and port: {address}")	1✔
150	if worker_address:	1✔
151	commands.append("--worker_address")	1✔
152	commands.append(worker_address)	1✔
153	logger.info(f"App Worker address and port: {worker_address}")	1✔
154
155	docker_run(	1✔
156	hostname,
157	map_name,
158	input_path,
159	output_path,
160	app_info,
161	pkg_info,
162	quiet,
163	commands,
164	health_check,
165	network,
166	nic,
167	use_all_nics,
168	gpus,
169	config,
170	render,
171	user,
172	terminal,
173	devices,
174	platform_config,
175	shared_memory_size,
176	args.uid == 0,
177	)
178
179
180	def _lookup_devices(devices: list[str]) -> list[str]:	1✔
181	"""
182	Looks up matching devices in /dev and returns a list
183	of fully qualified device paths.
184
185	Raises exception if any devices cannot be found.
186	"""
187	matched_devices = []	1✔
188	unmatched_devices = []	1✔
189
190	for dev in devices:	1✔
191	pattern = dev	1✔
192	if not pattern.startswith("/"):	1✔
193	pattern = "/dev/" + pattern	1✔
194	found_devices = glob(pattern)	1✔
195	if len(found_devices) == 0:	1✔
196	unmatched_devices.append(dev)	1✔
197	else:
198	matched_devices.extend(found_devices)	1✔
199
200	if len(unmatched_devices) > 0:	1✔
201	raise UnmatchedDeviceError(unmatched_devices)	1✔
202
203	return matched_devices	1✔
204
205
206	def _dependency_verification(map_name: str) -> bool:	1✔
207	"""Check if all the dependencies are installed or not.
208
209	Args:
210	map_name: HAP/MAP name
211
212	Returns:
213	True if all dependencies are satisfied, otherwise False.
214	"""
215	logger.info("Checking dependencies...")	1✔
216
217	# check for docker
218	prog = "docker"	1✔
219	logger.info(f'--> Verifying if "{prog}" is installed...\n')	1✔
220	if not shutil.which(prog):	1✔
221	logger.error(	1✔
222	'"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
223	prog,
224	)
225	return False	1✔
226
227	buildx_paths = [	1✔
228	os.path.expandvars("$HOME/.docker/cli-plugins"),
229	"/usr/local/lib/docker/cli-plugins",
230	"/usr/local/libexec/docker/cli-plugins",
231	"/usr/lib/docker/cli-plugins",
232	"/usr/libexec/docker/cli-plugins",
233	]
234	prog = "docker-buildx"	1✔
235	logger.info('--> Verifying if "%s" is installed...\n', prog)	1✔
236	buildx_found = False	1✔
237	for path in buildx_paths:	1✔
238	if shutil.which(os.path.join(path, prog)):	1✔
239	buildx_found = True	1✔
240
241	if not buildx_found:	1✔
242	logger.error(	1✔
243	'"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
244	prog,
245	)
246	return False	1✔
247
248	# check for map image
249	logger.info('--> Verifying if "%s" is available...\n', map_name)	1✔
250	if not image_exists(map_name):	1✔
251	logger.error("Unable to fetch required image.")	×
252	return False	×
253
254	return True	1✔
255
256
257	def _pkg_specific_dependency_verification(pkg_info: dict) -> bool:	1✔
258	"""Checks for any package specific dependencies.
259
260	Currently it verifies the following dependencies:
261	* If gpu has been requested by the application, verify that nvidia-ctk is installed.
262	Note: when running inside a Docker container, always assume nvidia-ctk is installed.
263	Args:
264	pkg_info: package manifest as a python dict
265
266	Returns:
267	True if all dependencies are satisfied, otherwise False.
268	"""
269	if os.path.exists("/.dockerenv"):	1✔
UNCOV 270	logger.info("--> Skipping nvidia-ctk check inside Docker...\n")	×
271	return True	×
272
273	requested_gpus = get_requested_gpus(pkg_info)	1✔
274	if requested_gpus > 0:	1✔
275	# check for NVIDIA Container TOolkit
276	prog = "nvidia-ctk"	1✔
277	logger.info('--> Verifying if "%s" is installed...\n', prog)	1✔
278	if not shutil.which(prog):	1✔
279	logger.error(	×
280	'"%s" not installed, please install NVIDIA Container Toolkit.', prog
281	)
282	return False	×
283
284	logger.info('--> Verifying "%s" version...\n', prog)	1✔
285	output = run_cmd_output(["nvidia-ctk", "--version"], "version")	1✔
286	match = re.search(r"([0-9]+\.[0-9]+\.[0-9]+)", output)	1✔
287	min_ctk_version = "1.12.0"	1✔
288	recommended_ctk_version = "1.14.1"	1✔
289
290	if match is None:	1✔
291	logger.error(	×
292	f"Error detecting NVIDIA Container Toolkit version. "
293	f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
294	)
295	return False	×
296	elif compare_versions(min_ctk_version, match.group()) > 0:	1✔
297	logger.error(	1✔
298	f"Found '{prog}' Version {match.group()}. "
299	f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
300	)
301	return False	1✔
302
303	return True	1✔
304
305
306	def execute_run_command(args: Namespace):	1✔
307	"""
308	Entrypoint for the Holoscan Run command.
309
310	Args:
311	args: Namespace object containing user arguments.
312
313	Returns:
314	None
315	"""
316	if not _dependency_verification(args.map):	1✔
317	logger.error("Execution Aborted")	1✔
318	sys.exit(2)	1✔
319	try:	1✔
320	# Fetch application manifest from MAP
321	app_info, pkg_info = _fetch_map_manifest(args.map)	1✔
322	except Exception as e:	1✔
323	logger.error(f"Failed to fetch MAP manifest: {e}")	1✔
324	sys.exit(2)	1✔
325
326	if not _pkg_specific_dependency_verification(pkg_info):	1✔
327	logger.error("Execution Aborted")	×
328	sys.exit(2)	×
329	try:	1✔
330	# Run Holoscan Application
331	_run_app(args, app_info, pkg_info)	1✔
332	except Exception as ex:	×
333	logger.debug(ex, exc_info=True)	×
334	logger.error(f"Error executing {args.map}: {ex}")	×
335	sys.exit(2)	×

nvidia-holoscan / holoscan-cli / 14097962944

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