fix(pytorch_xdit): key benchmark output dirs on the cluster target, not hostname

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')

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.

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_<cluster_target>_outputs``
+    - Multi-node: parses one ``flux_<cluster_target>_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")
 

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,