From 21e31bc93d32ac521c458b449c7510b0b829022f Mon Sep 17 00:00:00 2001 From: Pak Nin Lui Date: Wed, 28 Oct 2026 16:29:37 -0700 Subject: [PATCH 1/2] fix(health): fail TransferBench healthcheck test when TransferBench reports a failure test_transfer_bench_healthcheck only ran scan_test_results, whose patterns ("test FAIL ", "ABORT", ...) do not match TransferBench's healthcheck verdict lines such as "Testing all-to-all XGMI copies ........FAIL (56 test(s))". TransferBench's non-zero exit code was not checked either, and the Run Deck capture is best-effort, so a failing healthcheck was reported as a pytest pass. Run the healthcheck with orch.exec(detailed=True) and fail the test on a non-zero exit code. The healthcheck preset exits non-zero when any sub-test fails or when it cannot run (for example on an unsupported GPU model). Failed sub-tests are named from the "Testing ... FAIL (N test(s))" lines. Seen on MI300X with TransferBench v1.67.00 and v1.71.00: all-to-all XGMI copies failed on 56 of 56 pairs (exit code 2, also through the "sudo -n bash -c" wrapper), but the CVS test passed. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + .../unittests/test_transferbench_health.py | 59 +++++++++++++++++++ cvs/tests/health/transferbench_cvs.py | 28 ++++++++- 3 files changed, 85 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a711fcd38..4d871484f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Kernel-log health checks no longer echo each node's full dmesg or journal into the run log and HTML report. Each scan logs one summary line, and detected errors are still reported. - TorchTitan no longer writes the Hugging Face token into tokenizer-download or training commands. Each node reads `paths.hf_token_file` inside the container when that command runs. If that file is missing, empty, or not mounted, the stage fails and later stages are skipped. - Megatron and Primus no longer write the Hugging Face token into training, tokenizer-download, or job-launch commands. Each node reads `paths.hf_token_file` inside the container when that command runs. If that file is missing, empty, or not mounted, Megatron fails the test and the remaining training stages are skipped. Primus exports an empty `HF_TOKEN` and the run continues. +- The health TransferBench `healthcheck` test fails when TransferBench reports a failed sub-test or exits with an error. It previously passed, and only the Run Deck showed the failure. - ibperf: - Runs on nodes with fewer than 8 GPUs instead of failing with `KeyError: 'card4'`. - `verify_bw` now enforces `expected_results`; previously tests passed regardless of measured bandwidth. Configs with `verify_bw` set to `"True"` may now fail; re-baseline thresholds for your cluster. diff --git a/cvs/lib/unittests/test_transferbench_health.py b/cvs/lib/unittests/test_transferbench_health.py index e81aacf76..6ab0cbfdb 100644 --- a/cvs/lib/unittests/test_transferbench_health.py +++ b/cvs/lib/unittests/test_transferbench_health.py @@ -234,5 +234,64 @@ def test_missing_per_link_min_key_fails(self): self.assertIn('gpu_to_gpu_a2a_per_link_min', fail_test.call_args.args[0]) +HEALTHCHECK_ONE_FAIL = """\ +Testing HBM performance [READ] ........PASS +Testing bidirectional host<->device copies ........PASS +Testing all-to-all XGMI copies ........FAIL (56 test(s)) +""" + +HEALTHCHECK_ALL_PASS = HEALTHCHECK_ONE_FAIL.replace('FAIL (56 test(s))', 'PASS') + +# VERBOSE=1 prints the verdict on its own line, so no "Testing ....FAIL" line exists. +HEALTHCHECK_VERBOSE_FAIL = """\ +Testing all-to-all XGMI copies + GPU 00 to GPU 01: : Measured 40.17 Limit 43.65 +FAIL (56 test(s)) +""" + + +class TestParseTbHealthcheck(unittest.TestCase): + def test_all_pass_does_not_fail(self): + with patch.object(tb, 'fail_test') as fail_test: + tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_ALL_PASS, 'exit_code': 0}}) + fail_test.assert_not_called() + + def test_failed_subtest_is_named(self): + with patch.object(tb, 'fail_test') as fail_test: + tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_ONE_FAIL, 'exit_code': 2}}) + fail_test.assert_called_once() + message = fail_test.call_args.args[0] + self.assertIn('all-to-all XGMI copies', message) + self.assertIn('56 test(s)', message) + self.assertIn('nodeA', message) + + def test_nonzero_exit_without_fail_line_fails(self): + with patch.object(tb, 'fail_test') as fail_test: + tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_VERBOSE_FAIL, 'exit_code': 2}}) + fail_test.assert_called_once() + self.assertIn('exited with code 2', fail_test.call_args.args[0]) + + def test_generic_scan_misses_healthcheck_failure(self): + from cvs.lib import utils_lib + + with patch('cvs.lib.utils_lib.fail_test') as fail_test: + utils_lib.scan_test_results({'nodeA': HEALTHCHECK_ONE_FAIL}) + fail_test.assert_not_called() + + def test_run_transferbench_requests_detailed_results(self): + orch = MagicMock() + orch.orchestrator_type = 'baremetal' + orch.sudo_prefix.return_value = 'sudo -n ' + with patch.object(tb, 'detect_rocm_path', return_value='/opt/rocm'): + tb.run_transferbench( + orch, + {'path': '/tb', 'rocm_path': '/opt/rocm', 'num_cpu_devices': 2}, + 'healthcheck', + timeout=30, + detailed=True, + ) + self.assertTrue(orch.exec.call_args.kwargs['detailed']) + + if __name__ == '__main__': unittest.main() diff --git a/cvs/tests/health/transferbench_cvs.py b/cvs/tests/health/transferbench_cvs.py index ed968dae4..f47b3cb3e 100644 --- a/cvs/tests/health/transferbench_cvs.py +++ b/cvs/tests/health/transferbench_cvs.py @@ -184,7 +184,7 @@ def build_transferbench_command(path, rocm_path, preset, extra_env=None, sudo_pr return f'{sudo_prefix}bash -c {shlex.quote(inner)}' -def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecycle=None, stage=None): +def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecycle=None, stage=None, detailed=False): """Resolve ROCm/env and execute one TransferBench preset on all orch hosts.""" path = config_dict['path'] rocm_path = detect_rocm_path(orch, config_dict.get('rocm_path', '')) @@ -194,7 +194,7 @@ def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecy cmd = build_transferbench_command(path, rocm_path, preset, env, sudo_prefix=_payload_sudo_prefix(orch)) log.info('TransferBench command: %s', cmd) with timed_stage(lifecycle, stage or preset): - return orch.exec(cmd, timeout=timeout) + return orch.exec(cmd, timeout=timeout, detailed=detailed) def _tb_output_excerpt(out, tail=6000): @@ -462,6 +462,24 @@ def parse_tb_schmoo_bw(out_dict, exp_dict): ) +# e.g. "Testing all-to-all XGMI copies ........FAIL (56 test(s))" +_TB_HEALTHCHECK_FAIL_RE = re.compile(r'(?m)^[ \t]*Testing\s+(.+?)\s*\.{2,}\s*FAIL(?:\s*\((\d+)\s+test\(s\)\))?') + + +def parse_tb_healthcheck(results): + """Fail on a non-zero TransferBench healthcheck exit code, naming failed sub-tests from the output.""" + for node, res in results.items(): + output = res.get('output') or '' + failed = _TB_HEALTHCHECK_FAIL_RE.findall(output) + for name, count in failed: + fail_test(f"TransferBench healthcheck '{name.strip()}' FAIL ({count or '?'} test(s)) on node {node}") + if res.get('exit_code') != 0 and not failed: + fail_test( + f"TransferBench healthcheck exited with code {res.get('exit_code')} on node {node}. " + f"Output: {_tb_output_excerpt(output)}" + ) + + def _capture_tb_rundeck(res_dict, cluster_dict, recorder, *args): """Best-effort: a reporting problem must not change the TransferBench pass/fail.""" if res_dict is None: @@ -532,9 +550,13 @@ def test_transfer_bench_healthcheck( globals.error_list = [] log.info('Testcase Run TransferBench healthcheck') preset = config_dict.get('healthcheck_preset') or 'healthcheck' - out_dict = run_transferbench(orch, config_dict, preset, timeout=(60 * 3), lifecycle=lifecycle, stage='healthcheck') + results = run_transferbench( + orch, config_dict, preset, timeout=(60 * 3), lifecycle=lifecycle, stage='healthcheck', detailed=True + ) + out_dict = {node: res.get('output') or '' for node, res in results.items()} print_test_output(log, out_dict) scan_test_results(out_dict) + parse_tb_healthcheck(results) _capture_tb_rundeck( transferbench_res_dict, cluster_dict, From 0469ffb8b8f4969ee1a5af88240d990833953d76 Mon Sep 17 00:00:00 2001 From: Pak Nin Lui Date: Fri, 30 Oct 2026 16:36:07 -0700 Subject: [PATCH 2/2] fix(health): healthcheck handles sharded SSH and hosts missing from results Address review feedback on the TransferBench healthcheck exit-code check: - Pass print_console=False when run_transferbench requests detailed results. With SSH sharding (more than CVS_HOSTS_PER_SHARD hosts), the merged-output logger calls .splitlines() on each result and fails on the {output, exit_code} dicts. The healthcheck test logs the output itself through print_test_output. - Fail every expected host (orch.hosts) that has no result. Sharded runs send commands only to reachable hosts, so a host pruned during an earlier probe was absent from the results and the healthcheck passed without running there. Co-Authored-By: Claude Opus 5.5 --- cvs/lib/unittests/test_transferbench_health.py | 14 ++++++++++++++ cvs/tests/health/transferbench_cvs.py | 15 +++++++++++---- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/cvs/lib/unittests/test_transferbench_health.py b/cvs/lib/unittests/test_transferbench_health.py index 6ab0cbfdb..e2cf4bc0d 100644 --- a/cvs/lib/unittests/test_transferbench_health.py +++ b/cvs/lib/unittests/test_transferbench_health.py @@ -271,6 +271,19 @@ def test_nonzero_exit_without_fail_line_fails(self): fail_test.assert_called_once() self.assertIn('exited with code 2', fail_test.call_args.args[0]) + def test_expected_host_without_result_fails(self): + with patch.object(tb, 'fail_test') as fail_test: + tb.parse_tb_healthcheck( + {'nodeA': {'output': HEALTHCHECK_ALL_PASS, 'exit_code': 0}}, expected_hosts=['nodeA', 'nodeB'] + ) + fail_test.assert_called_once() + self.assertIn('no result for node nodeB', fail_test.call_args.args[0]) + + def test_no_results_fails_every_expected_host(self): + with patch.object(tb, 'fail_test') as fail_test: + tb.parse_tb_healthcheck({}, expected_hosts=['nodeA', 'nodeB']) + self.assertEqual(fail_test.call_count, 2) + def test_generic_scan_misses_healthcheck_failure(self): from cvs.lib import utils_lib @@ -291,6 +304,7 @@ def test_run_transferbench_requests_detailed_results(self): detailed=True, ) self.assertTrue(orch.exec.call_args.kwargs['detailed']) + self.assertFalse(orch.exec.call_args.kwargs['print_console']) if __name__ == '__main__': diff --git a/cvs/tests/health/transferbench_cvs.py b/cvs/tests/health/transferbench_cvs.py index f47b3cb3e..0735fedaf 100644 --- a/cvs/tests/health/transferbench_cvs.py +++ b/cvs/tests/health/transferbench_cvs.py @@ -194,7 +194,8 @@ def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecy cmd = build_transferbench_command(path, rocm_path, preset, env, sudo_prefix=_payload_sudo_prefix(orch)) log.info('TransferBench command: %s', cmd) with timed_stage(lifecycle, stage or preset): - return orch.exec(cmd, timeout=timeout, detailed=detailed) + # The sharded SSH handle can only print string results, so detailed callers log the output themselves. + return orch.exec(cmd, timeout=timeout, detailed=detailed, print_console=not detailed) def _tb_output_excerpt(out, tail=6000): @@ -466,8 +467,14 @@ def parse_tb_schmoo_bw(out_dict, exp_dict): _TB_HEALTHCHECK_FAIL_RE = re.compile(r'(?m)^[ \t]*Testing\s+(.+?)\s*\.{2,}\s*FAIL(?:\s*\((\d+)\s+test\(s\)\))?') -def parse_tb_healthcheck(results): - """Fail on a non-zero TransferBench healthcheck exit code, naming failed sub-tests from the output.""" +def parse_tb_healthcheck(results, expected_hosts=()): + """Fail on a non-zero TransferBench healthcheck exit code, naming failed sub-tests from the output. + + Sharded SSH runs omit hosts pruned as unreachable earlier, so every expected host must have a result. + """ + for node in expected_hosts: + if node not in results: + fail_test(f"TransferBench healthcheck returned no result for node {node}") for node, res in results.items(): output = res.get('output') or '' failed = _TB_HEALTHCHECK_FAIL_RE.findall(output) @@ -556,7 +563,7 @@ def test_transfer_bench_healthcheck( out_dict = {node: res.get('output') or '' for node, res in results.items()} print_test_output(log, out_dict) scan_test_results(out_dict) - parse_tb_healthcheck(results) + parse_tb_healthcheck(results, orch.hosts) _capture_tb_rundeck( transferbench_res_dict, cluster_dict,