feat(isvtest): provider-agnostic workload and conformance improvements

isvtest/src/isvtest/core/nvidia.py

@@ -61,9 +61,9 @@ def count_gpus_from_full_output(output: str) -> int:
     Returns:
         Number of GPUs found in output
 
-    Example pattern matched: "| 0  NVIDIA A100-SXM4-80GB"
+    Example pattern matched: "| 0  NVIDIA A100-SXM4-80GB", "| 0  Tesla T4"
     """
-    gpu_lines = re.findall(r"\|\s*\d+\s+NVIDIA", output, re.MULTILINE)
+    gpu_lines = re.findall(r"\|\s{1,4}\d{1,3}\s{2,}", output, re.MULTILINE)
     return len(gpu_lines)
 
 

isvtest/src/isvtest/validations/k8s_conformance.py

@@ -30,6 +30,7 @@
 
 from __future__ import annotations
 
+import os
 import tempfile
 import time
 import uuid
@@ -197,7 +198,32 @@ def run(self) -> None:
                 self.set_failed(completion_error, output=output[-4000:])
                 return
 
-            ok, junit_xml = self._exec_cat(namespace, self._POD_NAME, self._JUNIT_PATH)
+            # JUnit retrieval is multi-stage to tolerate flaky managed-K8s
+            # LoadBalancers that reset long-running TCP streams partway through
+            # a multi-megabyte `kubectl exec -- cat`:
+            #
+            #   1. If a provider has pre-staged the JUnit at the path in
+            #      $ISVTEST_CONFORMANCE_JUNIT_LOCAL_PATH (e.g. via a sidecar
+            #      copy through a different network path), use that. The
+            #      conformance pod stays Running until the harness cleans up,
+            #      so a pre-stage step that runs after `done` appears can
+            #      reliably exfiltrate the JUnit out-of-band.
+            #   2. Retry `kubectl exec -- cat` up to 3 times.
+            #   3. Fall back to `kubectl cp` (tar streaming) which uses
+            #      different stream framing than raw exec.
+            ok, junit_xml = self._try_local_junit()
+            if not ok or not junit_xml:
+                for attempt in range(1, 4):
+                    ok, junit_xml = self._exec_cat(namespace, self._POD_NAME, self._JUNIT_PATH, quiet=(attempt < 3))
+                    if ok and junit_xml:
+                        break
+                    self.log.warning(
+                        f"JUnit retrieval attempt {attempt}/3 via 'kubectl exec cat' failed; retrying in 5s"
+                    )
+                    time.sleep(5)
+            if not ok or not junit_xml:
+                # Final fallback: kubectl cp (tar streaming).
+                ok, junit_xml = self._kubectl_cp(namespace, self._POD_NAME, self._JUNIT_PATH)
             if not ok or not junit_xml:
                 logs = self._get_pod_logs(namespace, self._POD_NAME)
                 self.set_failed(
@@ -397,6 +423,57 @@ def _exec_cat(
             return False, ""
         return True, result.stdout
 
+    def _try_local_junit(self) -> tuple[bool, str]:
+        """Read a pre-staged JUnit XML from the local filesystem if the
+        provider has populated ``$ISVTEST_CONFORMANCE_JUNIT_LOCAL_PATH``.
+
+        Returns ``(ok, content)``. ``ok=False`` means either the env var
+        is unset, the file does not exist, or the file is empty.
+        """
+        local_path = os.environ.get("ISVTEST_CONFORMANCE_JUNIT_LOCAL_PATH", "")
+        if not local_path:
+            return False, ""
+        if not Path(local_path).is_file():
+            return False, ""
+        try:
+            with open(local_path, encoding="utf-8") as f:
+                content = f.read()
+        except OSError as exc:
+            self.log.warning(f"failed to read pre-staged JUnit at {local_path}: {exc}")
+            return False, ""
+        if not content:
+            return False, ""
+        self.log.info(f"Loaded pre-staged conformance JUnit from {local_path} ({len(content)} bytes)")
+        return True, content
+
+    def _kubectl_cp(self, namespace: str, pod_name: str, path: str) -> tuple[bool, str]:
+        """Copy ``path`` out of the pod to a temp file using `kubectl cp`, then
+        read it locally. Returns ``(ok, content)``.
+
+        `kubectl cp` uses tar streaming, which behaves differently from a raw
+        `kubectl exec -- cat` and recovers better from mid-stream TCP resets
+        on multi-megabyte files. Used as a fallback when the cat path fails.
+        """
+        with tempfile.NamedTemporaryFile(delete=False) as fh:
+            local_path = fh.name
+        try:
+            cmd = get_kubectl_base_shell("cp", f"{namespace}/{pod_name}:{path}", local_path)
+            result = self.run_command(cmd, timeout=self._EXEC_CAT_TIMEOUT)
+            if result.exit_code != 0:
+                self.log.warning(f"kubectl cp {path} failed: {result.stderr.strip()}")
+                return False, ""
+            try:
+                with open(local_path, encoding="utf-8") as f:
+                    content = f.read()
+            except OSError as exc:
+                self.log.warning(f"failed to read local copy of {path}: {exc}")
+                return False, ""
+            if not content:
+                return False, ""
+            return True, content
+        finally:
+            Path(local_path).unlink(missing_ok=True)
+
     def _get_pod_logs(self, namespace: str, pod_name: str) -> str:
         """Return the last 200 lines of pod logs, or ``""`` on failure."""
         cmd = get_kubectl_base_shell("logs", "-n", namespace, pod_name, "--tail=200")

isvtest/src/isvtest/workloads/k8s_nccl_multinode.py

@@ -237,6 +237,8 @@ def _patch_manifest(
         quick_mode: bool = False,
     ) -> str:
         """Replace placeholder values in the MPIJob manifest."""
+        import re as _re
+
         yaml_content = yaml_content.replace("name: nccl-allreduce-multinode", f"name: {job_name}", 1)
         yaml_content = yaml_content.replace("slotsPerWorker: 4", f"slotsPerWorker: {gpus_per_node}")
         yaml_content = yaml_content.replace("replicas: 2", f"replicas: {node_count}")
@@ -246,6 +248,36 @@ def _patch_manifest(
             yaml_content = yaml_content.replace("nvcr.io/nvidia/hpc-benchmarks:25.04", image)
         if quick_mode:
             yaml_content = yaml_content.replace("-b 8 -e 4G -f 2", "-b 1M -e 256M -f 2")
+
+        # Optional: remove runtimeClassName from worker pods.
+        # Set runtime_class_name: "" in config to omit runtimeClassName (e.g. GKE COS
+        # device-plugin model where the "nvidia" containerd handler is not configured).
+        runtime_class_name = self.config.get("runtime_class_name", None)
+        if runtime_class_name is not None:
+            if runtime_class_name == "":
+                yaml_content = _re.sub(r"\n          runtimeClassName: \S+", "", yaml_content)
+            else:
+                yaml_content = yaml_content.replace(
+                    "runtimeClassName: nvidia", f"runtimeClassName: {runtime_class_name}"
+                )
+
+        # Optional: remove the Launcher's control-plane nodeAffinity.
+        # The default manifest requires a node with node-role.kubernetes.io/control-plane,
+        # which is not present on clusters that do not expose control-plane nodes.
+        # Set override_launcher_affinity: true to drop the affinity and let the launcher
+        # schedule on any available node.
+        if self.config.get("override_launcher_affinity", False):
+            yaml_content = _re.sub(
+                r"\n          affinity:\n            nodeAffinity:\n"
+                r"              requiredDuringSchedulingIgnoredDuringExecution:\n"
+                r"                nodeSelectorTerms:\n"
+                r"                  - matchExpressions:\n"
+                r"                      - key: node-role\.kubernetes\.io/control-plane\n"
+                r"                        operator: Exists",
+                "",
+                yaml_content,
+            )
+
         return yaml_content
 
     def _resolve_compute_domain_mode(self) -> bool:

isvtest/src/isvtest/workloads/k8s_nim.py

@@ -19,6 +19,7 @@
 and runs basic inference validation.
 """
 
+import re
 import subprocess
 import uuid
 from pathlib import Path
@@ -51,6 +52,17 @@ def run(self) -> None:
         namespace = get_k8s_namespace()
         # Default timeout: 25 minutes (model download + load + inference)
         timeout = self.config.get("timeout", 1500)
+        # runtime_class_name: set "" to omit runtimeClassName (e.g. GKE COS device-plugin model).
+        # Default "nvidia" works for platforms with the NVIDIA GPU Operator RuntimeClass.
+        runtime_class_name = self.config.get("runtime_class_name", "nvidia")
+        # nim_memory_request / nim_memory_limit: tune to fit node allocatable memory.
+        # Default 16Gi/32Gi works for large instances; reduce for T4/smaller nodes.
+        nim_memory_request = self.config.get("nim_memory_request", "16Gi")
+        nim_memory_limit = self.config.get("nim_memory_limit", "32Gi")
+        # nim_max_model_len: caps vLLM KV-cache token budget.  On T4 16 GB the
+        # default max_seq_len of 131072 exceeds the available KV-cache (~54848
+        # tokens), causing the engine to abort.  Set e.g. "4096" for T4 nodes.
+        nim_max_model_len = self.config.get("nim_max_model_len", "")
 
         # Verify NGC secrets exist (using shared utility)
         success, error = ensure_ngc_secrets(namespace)
@@ -82,6 +94,24 @@ def run(self) -> None:
         # Replace job name
         yaml_content = yaml_content.replace("name: nim-llama-3b-inference-test", f"name: {job_name}")
 
+        # Apply runtimeClassName override: "" removes it, other values replace "nvidia".
+        if not runtime_class_name:
+            yaml_content = re.sub(r"\n      runtimeClassName: nvidia\n", "\n", yaml_content)
+        elif runtime_class_name != "nvidia":
+            yaml_content = yaml_content.replace("runtimeClassName: nvidia", f"runtimeClassName: {runtime_class_name}")
+
+        # Apply memory overrides for the NIM server container.
+        yaml_content = yaml_content.replace('memory: "16Gi"', f'memory: "{nim_memory_request}"')
+        yaml_content = yaml_content.replace('memory: "32Gi"', f'memory: "{nim_memory_limit}"')
+
+        # Inject NIM_MAX_MODEL_LEN when set (inserted before NIM_CACHE_PATH).
+        if nim_max_model_len:
+            inject = f'            - name: NIM_MAX_MODEL_LEN\n              value: "{nim_max_model_len}"\n'
+            yaml_content = yaml_content.replace(
+                "            - name: NIM_CACHE_PATH\n",
+                inject + "            - name: NIM_CACHE_PATH\n",
+            )
+
         self.log.info(f"Starting NIM inference workload (timeout: {timeout}s)")
         self.log.info("Steps: 1. Pull images, 2. Download model, 3. Load model (10-12m), 4. Run inference")
 

isvtest/src/isvtest/workloads/k8s_nim_helm.py

@@ -25,6 +25,7 @@
 
 import os
 import re
+import shlex
 import subprocess
 import time
 import uuid
@@ -39,6 +40,33 @@
 from isvtest.core.workload import BaseWorkloadCheck
 
 
+def _get_kubeconfig_from_kubectl() -> str:
+    """Extract the kubeconfig path from the KUBECTL env var, if present.
+
+    When KUBECTL is set to e.g. ``kubectl --kubeconfig=/path/to/config``,
+    this returns ``/path/to/config`` so that ``helm`` can be pointed at the
+    same cluster via ``--kubeconfig``.  Returns ``""`` when KUBECTL is not
+    set or does not contain a ``--kubeconfig`` flag.
+
+    Uses ``shlex.split`` so paths that legitimately contain spaces or quotes
+    survive the round-trip back to ``--kubeconfig`` callers.
+    """
+    kubectl = os.environ.get("KUBECTL", "")
+    if not kubectl:
+        return ""
+    try:
+        parts = shlex.split(kubectl)
+    except ValueError:
+        # Malformed KUBECTL (e.g. unbalanced quotes) - fall back to a naive split.
+        parts = kubectl.split()
+    for i, part in enumerate(parts):
+        if part.startswith("--kubeconfig="):
+            return part[len("--kubeconfig=") :]
+        if part == "--kubeconfig" and i + 1 < len(parts):
+            return parts[i + 1]
+    return ""
+
+
 @dataclass
 class NimPerfMetrics:
     """Performance metrics collected from GenAI-Perf."""
@@ -434,6 +462,34 @@ def _deploy_nim_helm(self, release_name: str, namespace: str, model: str, model_
             # Note: Don't use --wait here, _wait_for_nim_ready() handles waiting with progress logging
         ]
 
+        # Point helm at the cluster named by KUBECTL when it specifies a kubeconfig.
+        kubeconfig = _get_kubeconfig_from_kubectl()
+        if kubeconfig:
+            helm_cmd.extend(["--kubeconfig", kubeconfig])
+
+        # Optional memory overrides — useful when node allocatable RAM < chart defaults.
+        memory_request = self.config.get("memory_request", "")
+        memory_limit = self.config.get("memory_limit", "")
+        if memory_request:
+            helm_cmd.extend(["--set", f"resources.requests.memory={memory_request}"])
+        if memory_limit:
+            helm_cmd.extend(["--set", f"resources.limits.memory={memory_limit}"])
+
+        # Optional runtimeClassName override: set "" to omit (e.g. GKE COS device-plugin model).
+        # When not configured, the chart's own default applies.
+        runtime_class_name = self.config.get("runtime_class_name", None)
+        if runtime_class_name is not None:
+            helm_cmd.extend(["--set", f"runtimeClassName={runtime_class_name}"])
+
+        # Optional NIM_MAX_MODEL_LEN: caps the vLLM KV-cache token budget.
+        # On T4 16 GB the default max_seq_len (131072) exceeds the available
+        # KV-cache (~54848 tokens), causing the engine to abort at startup.
+        nim_max_model_len = self.config.get("nim_max_model_len", "")
+        if nim_max_model_len:
+            helm_cmd.extend(
+                ["--set", "env[0].name=NIM_MAX_MODEL_LEN", "--set-string", f"env[0].value={nim_max_model_len}"]
+            )
+
         try:
             result = subprocess.run(
                 helm_cmd,
@@ -804,15 +860,19 @@ def _dump_helm_status(self, release_name: str, namespace: str) -> None:
         self.log.error("Dumping Helm release status for debugging...")
 
         kubectl_base = get_kubectl_base_shell()
+        kubeconfig = _get_kubeconfig_from_kubectl()
+        kube_flag = f" --kubeconfig={shlex.quote(kubeconfig)}" if kubeconfig else ""
 
-        self.run_command(f"helm status {release_name} -n {namespace}")
+        self.run_command(f"helm status {release_name} -n {namespace}{kube_flag}")
         self.run_command(f"{kubectl_base} describe pods -n {namespace} -l app.kubernetes.io/instance={release_name}")
         self.run_command(f"{kubectl_base} logs -n {namespace} -l app.kubernetes.io/instance={release_name} --tail=100")
 
     def _cleanup_helm(self, release_name: str, namespace: str) -> None:
         """Clean up Helm release and downloaded chart file."""
         self.log.info(f"Cleaning up Helm release: {release_name}")
-        self.run_command(f"helm uninstall {release_name} -n {namespace} --wait=false")
+        kubeconfig = _get_kubeconfig_from_kubectl()
+        kube_flag = f" --kubeconfig={shlex.quote(kubeconfig)}" if kubeconfig else ""
+        self.run_command(f"helm uninstall {release_name} -n {namespace} --wait=false{kube_flag}")
 
         # Clean up downloaded chart file
         if self._chart_path:

isvtest/src/isvtest/workloads/k8s_stress.py

@@ -43,6 +43,11 @@ def run(self) -> None:
         memory_gb = self.config.get("memory_gb") or get_gpu_memory_gb()
         configured_gpu_count = self.config.get("gpu_count") or get_gpu_stress_gpu_count()
         cuda_arch = self.config.get("cuda_arch") or get_gpu_cuda_arch()
+        # runtime_class_name controls runtimeClassName in the pod spec.
+        # Default "nvidia" works for most platforms; set "" to omit (e.g. GKE
+        # COS where containerd uses the device-plugin model without a named
+        # nvidia runtime handler).
+        runtime_class_name = self.config.get("runtime_class_name", "nvidia")
 
         # Get GPU nodes
         # Note: We still rely on k8s_utils here for convenience
@@ -81,6 +86,7 @@ def run(self) -> None:
                 runtime=runtime,
                 memory_gb=memory_gb,
                 cuda_arch=cuda_arch,
+                runtime_class_name=runtime_class_name,
             )
 
             # Run the job (using run_k8s_job logic but adapted for pod since we create raw Pod here)
@@ -169,6 +175,7 @@ def _create_pod_yaml(
         runtime: int,
         memory_gb: int,
         cuda_arch: str | None = None,
+        runtime_class_name: str = "nvidia",
     ) -> str:
         """Create pod YAML for GPU stress test."""
         # Get the path to the gpu_stress_workload.py script
@@ -198,6 +205,7 @@ def _create_pod_yaml(
             env_vars.append('    - name: CUPY_CUDA_ARCH_LIST\n      value: "native"')
 
         env_section = "\n".join(env_vars)
+        runtime_class_line = f"  runtimeClassName: {runtime_class_name}\n" if runtime_class_name else ""
 
         pod_yaml = f"""apiVersion: v1
 kind: Pod
@@ -226,8 +234,7 @@ def _create_pod_yaml(
       limits:
         nvidia.com/gpu: "{gpu_count}"
     imagePullPolicy: IfNotPresent
-  runtimeClassName: nvidia
-  tolerations:
+{runtime_class_line}  tolerations:
   - key: "nvidia.com/gpu"
     operator: "Exists"
     effect: "NoSchedule"

isvtest/tests/test_k8s_conformance.py

@@ -19,6 +19,7 @@
 
 import json
 from collections.abc import Iterator
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -443,12 +444,36 @@ def test_pod_deleted_during_startup_fails_fast(self) -> None:
         assert len(phase_polls) <= 2
 
     def test_junit_retrieval_failure(self) -> None:
-        router = _happy_router({"cat /tmp/results/junit": fail(stderr="no such file")})
+        router = _happy_router(
+            {
+                "cat /tmp/results/junit": fail(stderr="no such file"),
+                " cp ": fail(stderr="tar: removing leading '/'"),
+            }
+        )
         check = _run_check(router)
 
         assert not check.passed
         assert "Could not retrieve" in check.message
 
+    def test_junit_retrieval_falls_back_to_kubectl_cp(self) -> None:
+        router = _happy_router({"cat /tmp/results/junit": fail(stderr="stream reset")})
+
+        def run(cmd: str, timeout: int | None = None) -> CommandResult:
+            if " cp " in cmd:
+                router.seen.append(cmd)
+                Path(cmd.split()[-1]).write_text(PASSING_JUNIT, encoding="utf-8")
+                return ok()
+            return router(cmd, timeout)
+
+        runner = MagicMock()
+        runner.run.side_effect = run
+        with patch("isvtest.validations.k8s_conformance.is_k8s_available", return_value=True):
+            check = _make_check(runner)
+            check.run()
+
+        assert check.passed
+        assert any(" cp " in cmd for cmd in router.seen)
+
     def test_malformed_junit_sets_failed(self) -> None:
         router = _happy_router({"cat /tmp/results/junit": ok("<<<not xml")})
         check = _run_check(router)