diff --git a/isvtest/src/isvtest/core/nvidia.py b/isvtest/src/isvtest/core/nvidia.py index 9482d437..60038464 100644 --- a/isvtest/src/isvtest/core/nvidia.py +++ b/isvtest/src/isvtest/core/nvidia.py @@ -61,9 +61,19 @@ def count_gpus_from_full_output(output: str) -> int: Returns: Number of GPUs found in output - Example pattern matched: "| 0 NVIDIA A100-SXM4-80GB" + Counts GPU identity rows by anchoring on the PCI Bus-Id that appears on + every physical GPU row of the table, regardless of vendor/name (e.g. + "NVIDIA A100-SXM4-80GB" or "Tesla T4"). Anchoring on the Bus-Id avoids + over-counting rows in the "Processes" table (which also start with a GPU + index but carry no Bus-Id) and MIG-device rows. + + Example row matched: "| 0 Tesla T4 Off | 00000000:00:04.0 Off | ... |" """ - gpu_lines = re.findall(r"\|\s*\d+\s+NVIDIA", output, re.MULTILINE) + gpu_lines = re.findall( + r"^\|\s+\d+\s+.+[0-9A-Fa-f]{8}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}\.\d", + output, + re.MULTILINE, + ) return len(gpu_lines) diff --git a/isvtest/src/isvtest/validations/k8s_conformance.py b/isvtest/src/isvtest/validations/k8s_conformance.py index 72645a2f..286fbaf8 100644 --- a/isvtest/src/isvtest/validations/k8s_conformance.py +++ b/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,36 @@ 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 + # Only log "retrying" and sleep when another exec attempt + # actually follows; on the last attempt fall straight + # through to the kubectl cp fallback below. + if attempt < 3: + 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 +427,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") diff --git a/isvtest/src/isvtest/workloads/k8s_nccl_multinode.py b/isvtest/src/isvtest/workloads/k8s_nccl_multinode.py index 3ecd8c76..a57866ce 100644 --- a/isvtest/src/isvtest/workloads/k8s_nccl_multinode.py +++ b/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: diff --git a/isvtest/src/isvtest/workloads/k8s_nim.py b/isvtest/src/isvtest/workloads/k8s_nim.py index 7b7b6431..5010780f 100644 --- a/isvtest/src/isvtest/workloads/k8s_nim.py +++ b/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,35 @@ 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 (GPU) container only. + # Anchor each substitution on the container's nvidia.com/gpu line inside + # its limits/requests block so the request and limit values can never + # clobber each other (e.g. request == old limit) or touch the sidecar. + yaml_content = re.sub( + r'(limits:\n\s+nvidia\.com/gpu: "1"\n\s+)memory: "32Gi"', + rf'\g<1>memory: "{nim_memory_limit}"', + yaml_content, + ) + yaml_content = re.sub( + r'(requests:\n\s+nvidia\.com/gpu: "1"\n\s+)memory: "16Gi"', + rf'\g<1>memory: "{nim_memory_request}"', + yaml_content, + ) + + # 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") diff --git a/isvtest/src/isvtest/workloads/k8s_nim_helm.py b/isvtest/src/isvtest/workloads/k8s_nim_helm.py index 1783bb1c..2c7a8088 100644 --- a/isvtest/src/isvtest/workloads/k8s_nim_helm.py +++ b/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: diff --git a/isvtest/src/isvtest/workloads/k8s_stress.py b/isvtest/src/isvtest/workloads/k8s_stress.py index b76a6c38..12e67dbc 100644 --- a/isvtest/src/isvtest/workloads/k8s_stress.py +++ b/isvtest/src/isvtest/workloads/k8s_stress.py @@ -35,6 +35,7 @@ class K8sGpuStressWorkload(BaseWorkloadCheck): description = "Run GPU stress test on all GPU nodes in the cluster." def run(self) -> None: + """Run the GPU stress workload on each GPU node and validate the result.""" # Get configuration namespace = get_k8s_namespace() image = self.config.get("image") or get_gpu_stress_image() @@ -43,6 +44,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 +87,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 +176,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 +206,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 +235,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" diff --git a/isvtest/tests/test_k8s_conformance.py b/isvtest/tests/test_k8s_conformance.py index 180ecf55..753fc08b 100644 --- a/isvtest/tests/test_k8s_conformance.py +++ b/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("<< None: + # "Tesla T4" has no "NVIDIA" prefix; it must still be counted. + assert count_gpus_from_full_output(_T4_SINGLE) == 1 + + def test_does_not_overcount_process_rows(self) -> None: + # Process-table rows also start with a GPU index but have no Bus-Id + # and must not be counted as additional GPUs. + assert count_gpus_from_full_output(_A100_TWO_WITH_PROCESSES) == 2 + + def test_no_gpus(self) -> None: + assert count_gpus_from_full_output("No devices were found") == 0 class TestParseCudaVersion: