diff --git a/tests/containers/kubernetes_utils.py b/tests/containers/kubernetes_utils.py index 60c2b62540..ddb86f1bb8 100644 --- a/tests/containers/kubernetes_utils.py +++ b/tests/containers/kubernetes_utils.py @@ -505,23 +505,83 @@ def check_not_none(value: Any, message: str) -> None: raise ValueError(message) +def _portforward_with_timeout( + core_v1_api: kubernetes.client.CoreV1Api, pod: kubernetes.client.models.V1Pod, timeout: float +) -> kubernetes.stream.ws_client.PortForward: + """Runs kubernetes.stream.portforward() with a wall-clock bound. + + The kubernetes client doesn't expose a per-call connect timeout for portforward() -- it calls + websocket.connect() with no timeout, so a wedged connection to the API server can hang this call + forever. The only alternative the library offers is a process-wide `websocket.setdefaulttimeout()`, + which isn't safe to use here since multiple SocketProxy instances (one per ImageDeployment) can be + making concurrent portforward() calls from different threads. So we run the call in a helper thread + and bound it with join(timeout) instead. + + If the call doesn't return in time, it is abandoned rather than joined: the daemon thread and any + partially established connection are leaked for the remainder of the test process. That's an + accepted tradeoff for test infrastructure that's about to raise TimeoutError and fail the test + (and the process will exit soon after) rather than hang forever with no recovery. + """ + pf_result: list[kubernetes.stream.ws_client.PortForward] = [] + error_result: list[Exception] = [] + + def _target() -> None: + try: + pf = kubernetes.stream.portforward( + api_method=core_v1_api.connect_get_namespaced_pod_portforward, + name=pod.metadata.name, + namespace=pod.metadata.namespace, + ports=",".join(str(p) for p in [8888]), + ) + pf_result.append(pf) + except Exception as e: # propagated to the caller below, not swallowed + error_result.append(e) + + thread = threading.Thread(target=_target, daemon=True) + thread.start() + thread.join(timeout) + if error_result: + raise error_result[0] + if not pf_result: + raise TimeoutError( + f"kubernetes.stream.portforward() to {pod.metadata.name} did not return within {timeout:.1f}s" + ) + return pf_result[0] + + @contextlib.contextmanager def exposing_contextmanager( - core_v1_api: kubernetes.client.CoreV1Api, pod: kubernetes.client.models.V1Pod + core_v1_api: kubernetes.client.CoreV1Api, + pod: kubernetes.client.models.V1Pod, + timeout: float = 30, ) -> Generator[socket]: # If we e.g., specify the wrong port, the pf = portforward() call succeeds, # but pf.connected will later flip to False # we need to check that _everything_ works before moving on - pf = None - s = None + # + # https://github.com/red-hat-data-services/notebooks/issues/2684: bound this retry loop (and each + # individual portforward() attempt via _portforward_with_timeout, since a single hung call would + # otherwise never let this loop's own deadline check run again) so a pod that never becomes + # reachable can't block the single-threaded SocketProxy forever and starve every later + # Wait.until retry. + deadline = time.monotonic() + timeout + pf: kubernetes.stream.ws_client.PortForward | None = None + s: kubernetes.stream.ws_client.PortForward._Port._Socket | socket.socket | None = None while not pf or not pf.connected or not s: - pf: kubernetes.stream.ws_client.PortForward = kubernetes.stream.portforward( - api_method=core_v1_api.connect_get_namespaced_pod_portforward, - name=pod.metadata.name, - namespace=pod.metadata.namespace, - ports=",".join(str(p) for p in [8888]), - ) - s: kubernetes.stream.ws_client.PortForward._Port._Socket | socket.socket | None = pf.socket(8888) + remaining = deadline - time.monotonic() + if remaining <= 0: + if s is not None: + s.close() + if pf is not None: + pf.close() + raise TimeoutError(f"Failed to establish a working portforward to {pod.metadata.name} within {timeout}s") + if s is not None: + s.close() + s = None + if pf is not None: + pf.close() + pf = _portforward_with_timeout(core_v1_api, pod, remaining) + s = pf.socket(8888) assert s, "Failed to establish connection" try: diff --git a/tests/containers/socket_proxy.py b/tests/containers/socket_proxy.py index 418c3979a1..de3a130eb9 100644 --- a/tests/containers/socket_proxy.py +++ b/tests/containers/socket_proxy.py @@ -94,10 +94,14 @@ def listen_and_serve_until_canceled(self): try: # handle client synchronously, which means that there can be at most one at a time self._handle_client(client_socket) - except (BrokenPipeError, ConnectionResetError) as e: + except (BrokenPipeError, ConnectionResetError, TimeoutError) as e: # BrokenPipeError happens when the proxy connects to the pod, but the service inside is not yet listening. + # TimeoutError is raised by remote_socket_factory() (e.g. exposing_contextmanager) when it can't + # establish a working connection within its own bound -- see https://github.com/red-hat-data-services/notebooks/issues/2684. # The client (Wait.until) will retry. - logging.info(f"Proxy connection to remote failed, likely due to service not being ready: {e}") + logging.info( + f"Proxy connection to remote failed, will retry on the next connection attempt: {e}" + ) # The client_socket is closed by the `with` statement in `_handle_client`. # We continue the loop to accept the next connection attempt. continue