diff --git a/kubeflow/spark/backends/kubernetes/backend.py b/kubeflow/spark/backends/kubernetes/backend.py index bdfc2adf5..202e7b227 100644 --- a/kubeflow/spark/backends/kubernetes/backend.py +++ b/kubeflow/spark/backends/kubernetes/backend.py @@ -457,13 +457,13 @@ def get_connect_url( if port is None: port_str = os.environ.get("SPARK_CONNECT_LOCAL_PORT") port = int(port_str) if port_str else random.randint(15002, 16002) - # Prefer pod when available (bypasses Service/EndpointSlice); then try svc names + # Prefer service-based port-forward (more stable via kube-proxy); fall back to pod candidates: list[tuple[str, str]] = [] + for svc in [info.service_name, f"{info.name}-server", f"{info.name}-svc"]: + if svc and not any(c[1] == svc for c in candidates): + candidates.append(("svc", svc)) if info.driver_pod_name: candidates.append(("pod", info.driver_pod_name)) - for svc in [f"{info.name}-svc", info.service_name, f"{info.name}-server"]: - if svc and not any(c[0] == "svc" and c[1] == svc for c in candidates): - candidates.append(("svc", svc)) seen: set[str] = set() for kind, target in candidates: key = f"{kind}/{target}" @@ -475,6 +475,8 @@ def get_connect_url( cmd = [ "kubectl", "port-forward", + "--address", + "127.0.0.1", key, f"{port}:{constants.SPARK_CONNECT_PORT}", "-n", @@ -596,9 +598,19 @@ def connect( while time.monotonic() - probe_start < delay_sec: # Check if port-forward process died if pf_proc is not None and pf_proc.poll() is not None: - logger.warning("Port-forward died during gRPC ready wait, restarting...") + stderr_b = pf_proc.stderr.read() if pf_proc.stderr else b"" + stderr_str = ( + stderr_b.decode("utf-8", errors="replace").strip() if stderr_b else "" + ) + logger.warning( + "Port-forward died during gRPC ready wait, restarting... " + "(exit=%s, stderr=%s)", + pf_proc.returncode, + stderr_str, + ) connect_url, pf_proc = self.get_connect_url(info, local_port=local_port) local_port = int(connect_url.split(":")[-1]) if pf_proc else None + time.sleep(2) # Allow new port-forward to stabilize # Verify port is still reachable if local_port and not self._wait_for_connect_port( "127.0.0.1", local_port, timeout_sec=1, interval_sec=0.5 diff --git a/kubeflow/spark/backends/kubernetes/backend_test.py b/kubeflow/spark/backends/kubernetes/backend_test.py index f954dc8c8..c465ad5a3 100644 --- a/kubeflow/spark/backends/kubernetes/backend_test.py +++ b/kubeflow/spark/backends/kubernetes/backend_test.py @@ -202,6 +202,7 @@ def mock_get_response(name: str) -> dict: server_status=models.SparkV1alpha1SparkConnectServerStatus( pod_name=f"{name}-0", pod_ip="10.0.0.5", + service_name=f"{name}-server-svc", ), ).to_dict() elif name == SPARK_CONNECT_PROVISIONING: @@ -725,12 +726,19 @@ def test_get_connect_url(kubernetes_backend, test_case): patch( "kubeflow.spark.backends.kubernetes.backend.subprocess.Popen", return_value=mock_popen, - ), + ) as mock_popen_cls, patch("kubeflow.spark.backends.kubernetes.backend.time.sleep"), patch.object(kubernetes_backend, "_wait_for_connect_port", return_value=True), ): url, proc = kubernetes_backend.get_connect_url(info) + # Verify service is tried before pod (service-first ordering) + first_call_args = mock_popen_cls.call_args[0][0] + first_call_key = first_call_args[4] + assert first_call_key.startswith("svc/"), ( + f"Expected service-first port-forward, got {first_call_key}" + ) + if "url_contains" in test_case.expected_output: assert test_case.expected_output["url_contains"] in url else: