fix(ci): recover deploy tests from vcluster tunnel drops

.github/actions/connect-vcluster/action.yml

@@ -49,7 +49,8 @@ runs:
         KUBECONFIG_TMP="${KUBECONFIG_PATH}.tmp"
 
         start_port_forward() {
-          kubectl port-forward -n "${NAMESPACE}" "svc/${VCLUSTER_NAME}" 8443:443 > "${PF_LOG}" 2>&1 &
+          printf '\n[%s] Starting vCluster API port-forward\n' "$(date -u +%FT%TZ)" >> "${PF_LOG}"
+          kubectl port-forward -n "${NAMESPACE}" "svc/${VCLUSTER_NAME}" 8443:443 >> "${PF_LOG}" 2>&1 &
           PF_PID=$!
           echo "${PF_PID}" > "${PF_PID_FILE}"
           echo "Started vCluster API port-forward pid ${PF_PID}; log: ${PF_LOG}"
@@ -143,7 +144,8 @@ runs:
             vcluster_name="$4"
 
             start_pf() {
-              kubectl port-forward -n "${namespace}" "svc/${vcluster_name}" 8443:443 > "${pf_log}" 2>&1 &
+              printf "\n[%s] Watchdog restarting vCluster API port-forward\n" "$(date -u +%FT%TZ)" >> "${pf_log}"
+              kubectl port-forward -n "${namespace}" "svc/${vcluster_name}" 8443:443 >> "${pf_log}" 2>&1 &
               echo "$!" > "${pf_pid_file}"
             }
 
@@ -159,7 +161,7 @@ runs:
               fi
               sleep 5
             done
-          ' _ "${PF_PID_FILE}" "${PF_LOG}" "${NAMESPACE}" "${VCLUSTER_NAME}" > "${WATCHDOG_LOG}" 2>&1 &
+          ' _ "${PF_PID_FILE}" "${WATCHDOG_LOG}" "${NAMESPACE}" "${VCLUSTER_NAME}" > "${WATCHDOG_LOG}" 2>&1 &
           echo "$!" > "${WATCHDOG_PID_FILE}"
           echo "Started vCluster API port-forward watchdog; log: ${WATCHDOG_LOG}"
         fi

.github/actions/dynamo-deploy-test/action.yml

@@ -158,6 +158,32 @@ runs:
         fi
 
         mkdir -p "${DYN_TEST_OUTPUT_PATH}"
+        for diagnostic_file in \
+          "${{ github.workspace }}/.vcluster-port-forward.log" \
+          "${{ github.workspace }}/.vcluster-port-forward-watchdog.log"; do
+          if [ -f "${diagnostic_file}" ]; then
+            cp "${diagnostic_file}" "${DYN_TEST_OUTPUT_PATH}/"
+          fi
+        done
+
+        {
+          echo "captured_at=$(date -u +%FT%TZ)"
+          curl -skS --connect-timeout 1 --max-time 2 \
+            -o /dev/null -w 'healthz_http_code=%{http_code}\n' \
+            https://127.0.0.1:8443/healthz || echo "healthz_unreachable=true"
+          for pid_file in \
+            "${{ github.workspace }}/.vcluster-port-forward.pid" \
+            "${{ github.workspace }}/.vcluster-port-forward-watchdog.pid"; do
+            if [ -f "${pid_file}" ]; then
+              pid="$(cat "${pid_file}")"
+              echo "$(basename "${pid_file}")=${pid}"
+              ps -p "${pid}" -o pid=,ppid=,stat=,etime=,command= || true
+            else
+              echo "$(basename "${pid_file}")=missing"
+            fi
+          done
+        } > "${DYN_TEST_OUTPUT_PATH}/vcluster-port-forward-status.log" 2>&1
+
         kubectl cluster-info dump \
           --namespaces="${namespaces}" \
           --output-directory="${DYN_TEST_OUTPUT_PATH}/cluster-info" || true

tests/deploy/dgd_utils.py

@@ -12,6 +12,8 @@
 from dataclasses import dataclass, field
 from typing import Any, List, Literal, Optional
 
+import aiohttp
+import httpx
 import kr8s
 import pytest
 import requests
@@ -20,6 +22,7 @@
 from kubernetes_asyncio import client, config
 from kubernetes_asyncio.client import exceptions
 
+from tests.utils.client import send_request
 from tests.utils.test_output import resolve_test_output_path
 
 logger = logging.getLogger(__name__)
@@ -49,6 +52,8 @@
 # Minimum response content length to validate that the model is generating meaningful output.
 # This matches the validation threshold from the original shell-based deployment tests.
 MIN_RESPONSE_CONTENT_LENGTH = 100
+VCLUSTER_CONNECTION_RETRY_LIMIT = 3
+VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS = 5
 
 
 def validate_chat_response(
@@ -1472,28 +1477,42 @@ def get_service(self, service_name=None):
         return Service.get(full_service_name, namespace=self.namespace)
 
     def get_pods(self, service_names: list[str] | None = None) -> dict[str, list[Pod]]:
-        result: dict[str, list[Pod]] = {}
-
         if not service_names:
             service_names = [service.name for service in self.deployment_spec.services]
 
-        for original_name in service_names:
-            # List pods using stable labels that are not affected by worker hash suffixes.
-            label_selector = (
-                f"nvidia.com/dynamo-graph-deployment-name={self._deployment_name},"
-                f"nvidia.com/dynamo-component={original_name}"
-            )
-
-            pods: list[Pod] = []
-
-            for pod in kr8s.get(
-                "pods", namespace=self.namespace, label_selector=label_selector
-            ):
-                pods.append(pod)  # type: ignore[arg-type]
+        for attempt in range(VCLUSTER_CONNECTION_RETRY_LIMIT + 1):
+            try:
+                result: dict[str, list[Pod]] = {}
+                for original_name in service_names:
+                    # List pods using stable labels that are not affected by worker hash suffixes.
+                    label_selector = (
+                        f"nvidia.com/dynamo-graph-deployment-name={self._deployment_name},"
+                        f"nvidia.com/dynamo-component={original_name}"
+                    )
 
-            result[original_name] = pods
+                    result[original_name] = list(  # type: ignore[arg-type]
+                        kr8s.get(
+                            "pods",
+                            namespace=self.namespace,
+                            label_selector=label_selector,
+                        )
+                    )
+                return result
+            except httpx.TransportError as error:
+                if attempt == VCLUSTER_CONNECTION_RETRY_LIMIT:
+                    raise
+                self._logger.warning(
+                    "vCluster API connection failed while listing pods; "
+                    "the port-forward watchdog may restore the tunnel, "
+                    "retrying in %ss (%s/%s): %s",
+                    VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS,
+                    attempt + 1,
+                    VCLUSTER_CONNECTION_RETRY_LIMIT,
+                    error,
+                )
+                time.sleep(VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS)
 
-        return result
+        raise AssertionError("unreachable")
 
     def get_pod_manifest_logs_metrics(self, service_name: str, pod: Pod, suffix=""):
         directory = os.path.join(self.log_dir, service_name)
@@ -1614,18 +1633,37 @@ async def _delete_deployment(self):
         """
         Delete the DynamoGraphDeployment CR.
         """
-        try:
-            if self._deployment_name and self._custom_api is not None:
+        if not self._deployment_name or self._custom_api is None:
+            return
+
+        for attempt in range(VCLUSTER_CONNECTION_RETRY_LIMIT + 1):
+            try:
                 await self._custom_api.delete_namespaced_custom_object(
                     group="nvidia.com",
                     version=self.deployment_spec.api_version,
                     namespace=self.namespace,
                     plural="dynamographdeployments",
                     name=self._deployment_name,
                 )
-        except exceptions.ApiException as e:
-            if e.status != 404:  # Ignore if already deleted
+                return
+            except exceptions.ApiException as error:
+                if error.status == 404:  # Ignore if already deleted
+                    return
                 raise
+            except aiohttp.ClientConnectionError as error:
+                if attempt == VCLUSTER_CONNECTION_RETRY_LIMIT:
+                    raise
+                self._logger.warning(
+                    "vCluster API connection failed while deleting deployment %s; "
+                    "the port-forward watchdog may restore the tunnel, "
+                    "retrying in %ss (%s/%s): %s",
+                    self._deployment_name,
+                    VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS,
+                    attempt + 1,
+                    VCLUSTER_CONNECTION_RETRY_LIMIT,
+                    error,
+                )
+                await asyncio.sleep(VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS)
 
     def port_forward(
         self, pod: Pod, remote_port: int, max_connection_attempts: int = 3
@@ -1722,6 +1760,57 @@ def port_forward(
             )
             return None
 
+    def send_request_with_port_forward_retry(
+        self,
+        pod: Pod,
+        remote_port: int,
+        endpoint: str,
+        payload: dict[str, Any],
+        timeout: float,
+        port_forward: Any,
+        request_sender: Any = send_request,
+    ) -> requests.Response:
+        """Retry one request after rebuilding a dropped pod port-forward."""
+        active_port_forward = port_forward
+
+        for attempt in range(2):
+            url = f"http://localhost:{active_port_forward.local_port}{endpoint}"
+            try:
+                return request_sender(url, payload, timeout=timeout, method="POST")
+            except (
+                requests.ConnectionError,
+                requests.Timeout,
+                httpx.TransportError,
+            ) as error:
+                if attempt == 1:
+                    raise
+
+                self._logger.warning(
+                    "Frontend request transport failed; rebuilding the pod "
+                    "port-forward and retrying once: %s",
+                    error,
+                )
+                try:
+                    active_port_forward.stop()
+                except RuntimeError as stop_error:
+                    if "anext()" not in str(
+                        stop_error
+                    ) and "already running" not in str(stop_error):
+                        raise
+                    self._logger.debug(
+                        "Ignoring expected error while stopping failed frontend "
+                        "port-forward: %s",
+                        stop_error,
+                    )
+
+                time.sleep(VCLUSTER_CONNECTION_RETRY_DELAY_SECONDS)
+                replacement = self.port_forward(pod, remote_port)
+                if replacement is None:
+                    raise
+                active_port_forward = replacement
+
+        raise AssertionError("unreachable")
+
     async def _cleanup(self):
         try:
             # Collect logs/metrics first; any PFs opened here will be tracked and stopped below.
@@ -1764,13 +1853,30 @@ async def __aenter__(self):
             await self._create_deployment()
             await self._wait_for_ready(timeout=self.readiness_timeout)
 
-        except:
-            await self._cleanup()
+        except BaseException:
+            try:
+                await self._cleanup()
+            except (aiohttp.ClientConnectionError, httpx.TransportError):
+                self._logger.exception(
+                    "vCluster connection failed during cleanup after deployment "
+                    "setup failure; preserving the original error"
+                )
             raise
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self._cleanup()
+        if exc_type is None:
+            await self._cleanup()
+            return None
+
+        try:
+            await self._cleanup()
+        except (aiohttp.ClientConnectionError, httpx.TransportError):
+            self._logger.exception(
+                "vCluster connection failed during cleanup after test failure; "
+                "preserving the original error"
+            )
+        return False
 
 
 async def main():