diff --git a/cvs/lib/unittests/test_utils_lib.py b/cvs/lib/unittests/test_utils_lib.py index 39c03c755..aed40273e 100644 --- a/cvs/lib/unittests/test_utils_lib.py +++ b/cvs/lib/unittests/test_utils_lib.py @@ -19,6 +19,11 @@ def test_scan_test_results_no_failure(self, mock_fail_test): utils_lib.scan_test_results(out_dict) mock_fail_test.assert_not_called() + def test_cluster_target_output_label_strips_and_sanitizes(self): + self.assertEqual(utils_lib.cluster_target_output_label(" node1.example.com "), "node1.example.com") + self.assertEqual(utils_lib.cluster_target_output_label("a/b"), "a_b") + self.assertEqual(utils_lib.cluster_target_output_label(""), "unknown_node") + def test_get_model_from_rocm_smi_output_matches_marketing_name(self): self.assertEqual(utils_lib.get_model_from_rocm_smi_output('Card series: AMD Instinct MI300X'), 'mi300x') self.assertEqual(utils_lib.get_model_from_rocm_smi_output('Card series: AMD Instinct MI325X'), 'mi325') diff --git a/cvs/lib/utils_lib.py b/cvs/lib/utils_lib.py index eef483f28..fea887693 100644 --- a/cvs/lib/utils_lib.py +++ b/cvs/lib/utils_lib.py @@ -435,6 +435,19 @@ def resolve_test_config_placeholders(config_dict, cluster_dict): return resolved_config +def cluster_target_output_label(cluster_node_key: str) -> str: + """ + Stable label for per-node benchmark output directories. + + Uses the cluster ``node_dict`` key (the SSH target string) instead of the remote + ``hostname`` command, so the same node does not produce multiple output trees when + short and fully-qualified hostnames vary across environments. + """ + if not cluster_node_key: + return "unknown_node" + return str(cluster_node_key).strip().replace("/", "_") + + def collect_system_metadata(phdl, cluster_dict, config_dict, test_command=None, env_vars=None): """ Collect comprehensive system metadata from compute nodes for test reporting. diff --git a/cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_single.py b/cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_single.py index 0f8732e09..48bf15a66 100644 --- a/cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_single.py +++ b/cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_single.py @@ -17,6 +17,7 @@ from cvs.lib.parallel_ssh_lib import Pssh from cvs.lib.utils_lib import ( + cluster_target_output_label, fail_test, update_test_result, get_model_from_rocm_smi_output, @@ -559,10 +560,12 @@ def test_run_flux1_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t use_torch_compile = flux_params['use_torch_compile'] torchrun_nproc = flux_params['torchrun_nproc'] - # Get hostnames from all nodes - log.info(f"Getting hostnames from {len(s_phdl.host_list)} node(s)") - hostname_result = s_phdl.exec('hostname') - node_to_hostname = {node: hostname_result[node].strip() for node in s_phdl.host_list} + log.info(f"Resolving output directory labels for {len(s_phdl.host_list)} node(s)") + hostname_result = s_phdl.exec('hostname', print_console=False) + node_to_hostname = {node: (hostname_result.get(node, "") or "").strip() or node for node in s_phdl.host_list} + node_to_out_label = {node: cluster_target_output_label(node) for node in s_phdl.host_list} + for node in s_phdl.host_list: + log.info(f"Node {node}: output label '{node_to_out_label[node]}' (hostname: {node_to_hostname[node]})") # Build common docker command components device_list = inference_dict['container_config']['device_list'] @@ -611,8 +614,8 @@ def test_run_flux1_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t docker_cmds = [] for node in s_phdl.host_list: - hostname = node_to_hostname[node] - output_dir = f"{output_base_dir}/flux_{hostname}_outputs" + out_label = node_to_out_label[node] + output_dir = f"{output_base_dir}/flux_{out_label}_outputs" # Create output directory command mkdir_cmds.append(f"mkdir -p {output_dir}") @@ -646,7 +649,7 @@ def test_run_flux1_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t f"{torchrun_cmd}" ) docker_cmds.append(docker_cmd) - log.info(f"Node {node} ({hostname}) will write to: {output_dir}") + log.info(f"Node {node} will write to: {output_dir}") # Create output directories on all nodes in parallel log.info(f"Creating output directories on {len(s_phdl.host_list)} node(s)") @@ -688,7 +691,7 @@ def test_run_flux1_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t # picking up stale outputs from previous runs on other hosts. if len(getattr(s_phdl, "host_list", []) or []) == 1: only_node = s_phdl.host_list[0] - inference_dict["_test_output_dir"] = f"{output_base_dir}/flux_{node_to_hostname[only_node]}_outputs" + inference_dict["_test_output_dir"] = f"{output_base_dir}/flux_{node_to_out_label[only_node]}_outputs" update_test_result() @@ -698,8 +701,8 @@ def test_parse_and_validate_results(s_phdl, inference_dict, benchmark_params_dic Parse benchmark outputs and validate against thresholds. Handles both single-node and multi-node runs: - - Single node: parses flux_{hostname}_outputs - - Multi-node: parses all flux_*_outputs directories and validates each + - Single node: parses ``flux__outputs`` + - Multi-node: parses one ``flux__outputs`` per node and validates each Uses FluxOutputParser to: - Locate results/timing.json @@ -726,28 +729,21 @@ def test_parse_and_validate_results(s_phdl, inference_dict, benchmark_params_dic if inference_dict.get("_test_output_dir"): output_dirs = [inference_dict["_test_output_dir"]] else: - # Otherwise, derive expected output dirs from the current nodes' hostnames. + # Derive from cluster SSH targets (same labels as the run step). try: - head_node = s_phdl.host_list[0] - hostname_out = s_phdl.exec('hostname', print_console=False) - expected_hostnames = [] - for node in s_phdl.host_list: - hn = (hostname_out.get(node, "") or "").strip() or node - expected_hostnames.append(hn) + expected_labels = [cluster_target_output_label(n) for n in s_phdl.host_list] except Exception: - # Fallback to head node only - expected_hostnames = [head_node] if 'head_node' in locals() else [] + expected_labels = [] - if not expected_hostnames: - fail_test("Could not determine node hostnames to locate Flux outputs") + if not expected_labels: + fail_test("Could not determine cluster node keys to locate Flux outputs") update_test_result() return - # Single-node: parse only that node's directory. if node_count <= 1: - output_dirs = [f"{output_base_dir}/flux_{expected_hostnames[0]}_outputs"] + output_dirs = [f"{output_base_dir}/flux_{expected_labels[0]}_outputs"] else: - output_dirs = [f"{output_base_dir}/flux_{hn}_outputs" for hn in expected_hostnames] + output_dirs = [f"{output_base_dir}/flux_{lab}_outputs" for lab in expected_labels] log.info(f"Found {len(output_dirs)} output directory(ies) to parse") diff --git a/cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_single.py b/cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_single.py index f3b611a12..ce94d0a9e 100644 --- a/cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_single.py +++ b/cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_single.py @@ -17,6 +17,7 @@ from cvs.lib.parallel_ssh_lib import Pssh from cvs.lib.utils_lib import ( + cluster_target_output_label, fail_test, update_test_result, get_model_from_rocm_smi_output, @@ -421,10 +422,13 @@ def test_run_wan22_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t compile_flag = "--compile" if wan_params['compile'] else "" torchrun_nproc = wan_params['torchrun_nproc'] - # Get hostnames from all nodes - log.info(f"Getting hostnames from {len(s_phdl.host_list)} node(s)") - hostname_result = s_phdl.exec('hostname') - node_to_hostname = {node: hostname_result[node].strip() for node in s_phdl.host_list} + # Output dirs use cluster SSH targets (node_dict keys), not `hostname`, to avoid FQDN drift. + log.info(f"Resolving output directory labels for {len(s_phdl.host_list)} node(s)") + hostname_result = s_phdl.exec('hostname', print_console=False) + node_to_hostname = {node: (hostname_result.get(node, "") or "").strip() or node for node in s_phdl.host_list} + node_to_out_label = {node: cluster_target_output_label(node) for node in s_phdl.host_list} + for node in s_phdl.host_list: + log.info(f"Node {node}: output label '{node_to_out_label[node]}' (hostname: {node_to_hostname[node]})") # Prefer the resolved checkpoint dir computed in test_verify_hf_cache_or_download. ckpt_dir = inference_dict.get("_resolved_ckpt_dir_container") @@ -475,8 +479,8 @@ def test_run_wan22_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t docker_cmds = [] for node in s_phdl.host_list: - hostname = node_to_hostname[node] - output_dir = f"{output_base_dir}/wan_22_{hostname}_outputs" + out_label = node_to_out_label[node] + output_dir = f"{output_base_dir}/wan_22_{out_label}_outputs" outputs_dir = f"{output_dir}/outputs" # Create output directory command @@ -511,7 +515,7 @@ def test_run_wan22_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t f"{torchrun_cmd}" ) docker_cmds.append(docker_cmd) - log.info(f"Node {node} ({hostname}) will write to: {output_dir}") + log.info(f"Node {node} will write to: {output_dir}") # Create output directories on all nodes in parallel log.info(f"Creating output directories on {len(s_phdl.host_list)} node(s)") @@ -550,8 +554,10 @@ def test_run_wan22_benchmark(s_phdl, inference_dict, benchmark_params_dict, hf_t except Exception as e: fail_test(f"Benchmark execution failed with exception: {e}") - # Note: _test_output_dir is no longer set since we run on multiple nodes. - # The parsing test will use output_base_dir to find all wan_22_*_outputs directories. + # Single-node: pin the run output dir so parse does not depend on probing hostnames later. + if len(getattr(s_phdl, "host_list", []) or []) == 1: + only_node = s_phdl.host_list[0] + inference_dict["_test_output_dir"] = f"{output_base_dir}/wan_22_{node_to_out_label[only_node]}_outputs" update_test_result() @@ -571,15 +577,13 @@ def test_parse_and_validate_results(s_phdl, inference_dict, benchmark_params_dic output_dir = inference_dict.get('_test_output_dir') if not output_dir: - # Allow running this test standalone by deriving the output directory - # from the configured output_base_dir and current hostname. + # Standalone parse: derive from cluster target keys (same as run step). try: head_node = s_phdl.host_list[0] - hostname_out = s_phdl.exec('hostname', print_console=False) - hostname = hostname_out.get(head_node, '').strip() or head_node + out_label = cluster_target_output_label(head_node) output_base_dir = inference_dict.get('output_base_dir') if output_base_dir: - output_dir = f"{output_base_dir}/wan_22_{hostname}_outputs" + output_dir = f"{output_base_dir}/wan_22_{out_label}_outputs" log.info(f"Derived output directory: {output_dir}") except Exception: output_dir = None @@ -601,15 +605,7 @@ def test_parse_and_validate_results(s_phdl, inference_dict, benchmark_params_dic agg, agg_errors = None, [] if base_dir and node_count > 1: # Filter aggregation to the current nodes only (avoid mixing with stale dirs). - try: - hostnames = s_phdl.exec("hostname", print_console=False) - expected_dirnames = [] - for _, hn in (hostnames or {}).items(): - h = (hn or "").strip() - if h: - expected_dirnames.append(f"wan_22_{h}_outputs") - except Exception: - expected_dirnames = [] + expected_dirnames = [f"wan_22_{cluster_target_output_label(n)}_outputs" for n in s_phdl.host_list] agg, agg_errors = WanOutputParser.parse_runs_under_base_dir( base_dir=base_dir,