diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py index d6f7ec8659..4d4462fdb9 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py @@ -608,7 +608,11 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate return map_docker_state_to_starting(container_id, state) if state == "running": - host_url = self._primary_host_url(host_ports) + # The default (no-declared-probe) reachability check TCP-connects the port, so + # it must see only TCP mappings: a UDP-only workload has no TCP listener and + # would otherwise be gated STARTING forever. Endpoints still carry every port. + tcp_host_ports = self._extract_host_ports(container, protocol="tcp") + host_url = self._primary_host_url(tcp_host_ports) config = await self._load_config_from_labels(workspace, labels) probe = None if config is not None and config.containers: @@ -617,7 +621,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate container=container, probe=probe, host_url=host_url, - host_ports=host_ports, + host_ports=tcp_host_ports, ) if ready and restart_policy == "Always": sidecar_ok, sidecar_reason = await self._sidecars_healthy(workspace, name, config) @@ -1088,13 +1092,21 @@ async def _load_config_for_deployment_entity( except Exception: return None - def _extract_host_ports(self, container: DockerContainer) -> dict[int, int]: + def _extract_host_ports(self, container: DockerContainer, *, protocol: str | None = None) -> dict[int, int]: + """Map container port -> published host port. + + With *protocol* (e.g. ``"tcp"``) only mappings of that protocol are returned; + docker keys the port map as ``"/"``. Defaults to every protocol. + """ result: dict[int, int] = {} ports = container.ports or {} for key, bindings in ports.items(): if not bindings: continue - container_port = int(str(key).split("/")[0]) + key_str = str(key) + if protocol is not None and not key_str.endswith(f"/{protocol}"): + continue + container_port = int(key_str.split("/")[0]) host_port = bindings[0].get("HostPort") if host_port: result[container_port] = int(host_port) diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/probes.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/probes.py index 9aa97511a7..9a45cba579 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/probes.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/probes.py @@ -16,6 +16,11 @@ logger = logging.getLogger(__name__) +# Timeout for the default reachability probe used when a container declares no +# readinessProbe. Keep it well under the reconciler poll interval so a probe never +# stalls the reconcile loop. +_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS = 2.0 + async def check_readiness_probe( *, @@ -25,9 +30,17 @@ async def check_readiness_probe( host_ports: dict[int, int] | None = None, named_ports: dict[str, int] | None = None, ) -> tuple[bool, str]: - """Return (ready, reason). When no probe is configured, running implies ready.""" + """Return (ready, reason). + + With a declared probe, evaluate it. With no declared probe, a workload that + publishes a port is only ready once that port accepts a connection, so status does + not race the process's bind(); a portless workload has no socket to reach, so running + implies ready. + """ if probe is None: - return True, "no readiness probe configured" + if host_url is None or not host_ports: + return True, "no readiness probe configured" + return await _check_default_tcp(host_url) if probe.exec_action is not None and probe.exec_action.command: return await _check_exec_probe(container, probe) @@ -151,5 +164,24 @@ def _connect() -> None: return False, f"tcp probe failed: {exc}" +async def _check_default_tcp(host_url: str) -> tuple[bool, str]: + """Default reachability check: TCP-connect the primary published host port.""" + parsed = urlparse(host_url) + host = parsed.hostname or "127.0.0.1" + port = parsed.port + if port is None: + return True, "no host port to probe" + + def _connect() -> None: + with socket.create_connection((host, port), timeout=_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS): + return + + try: + await asyncio.wait_for(asyncio.to_thread(_connect), timeout=_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS) + return True, f"default tcp probe connected ({host}:{port})" + except Exception as exc: + return False, f"default tcp probe not ready ({host}:{port}): {exc}" + + def host_url_for_port(host: str, host_port: int, *, scheme: str = "http") -> str: return f"{scheme}://{host}:{host_port}" diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/openshell/backend.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/openshell/backend.py index b07f8364e8..88cfe774a3 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/openshell/backend.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/openshell/backend.py @@ -134,6 +134,29 @@ def _delivery_script(path: str, mode: int) -> str: """ _LOG_TAIL_LINES = 20 +# Timeout for the default (no-declared-probe) loopback reachability probe. A declared +# readinessProbe uses its own timeout_seconds instead. +_DEFAULT_READINESS_TIMEOUT_SECONDS = 3 + +# Headroom added to a network probe's own timeout when bounding the ExecSandbox RPC, so +# the RPC outlives the in-sandbox python timeout and captures its (non-zero) exit rather +# than being cut off first. +_READINESS_EXEC_TIMEOUT_MARGIN_SECONDS = 5 + +# Loopback readiness probe programs, run by the sandbox's python. urlopen raises on a +# refused connection or an HTTP status >= 400 (so a still-starting 503 reads as not +# ready); create_connection raises until the socket is actually bound. The HTTP probe +# uses an unverified TLS context so an https readinessProbe against a loopback/self-signed +# cert is not rejected on verification (matching Kubernetes httpGet HTTPS probe semantics); +# the context is ignored for plain http, so one program covers both schemes. +_HTTP_PROBE_PROGRAM = ( + "import sys, ssl, urllib.request; " + "urllib.request.urlopen(sys.argv[1], timeout=float(sys.argv[2]), context=ssl._create_unverified_context())" +) +_TCP_PROBE_PROGRAM = ( + "import sys, socket; socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=float(sys.argv[3])).close()" +) + # Cached on first status read; needs the proto enums so it cannot be built at import # time (see _ensure_openshell). None until built. _PHASE_TO_STATUS: dict[int, DeploymentStatus] | None = None @@ -544,6 +567,13 @@ async def _advance_provisioning(self, sandbox: Any, sandbox_nm: str, workspace: if state == "pending": return BackendStatusUpdate(status="STARTING", status_message="Serve launched; awaiting serve pid") + # A live pid has bound its pidfile, not necessarily its socket. Do not expose + # (which reads as READY) until the workload actually accepts a connection, so a + # caller trusting READY does not 502 against a process still starting up. + pending = await self._readiness_pending(sandbox_id, container) + if pending is not None: + return pending + # Serve launched but ports not yet exposed: expose them. try: endpoints = await self._expose_ports(sandbox_nm, container) @@ -658,16 +688,21 @@ async def _try_get_sandbox(self, sandbox_nm: str) -> Any | None: return response.sandbox async def _exec_detached( - self, sandbox_id: str, command: list[str], *, stdin: bytes | None = None + self, sandbox_id: str, command: list[str], *, timeout: int | None = None, stdin: bytes | None = None ) -> tuple[int | None, str]: - """Run a command, draining its event stream. Returns (exit_code, combined output). + """Run *command* to completion, returning (exit_code, stdout+stderr merged); + ``exit_code`` is None when the stream carried no exit event. + + *timeout* bounds both the RPC and the sandbox-side command, defaulting to the + executor's control-plane ``request_timeout_seconds``; readiness probes pass a much + shorter bound so a hung probe cannot stall the serial reconcile loop. When *stdin* is given it is streamed to the command as its standard input. The ExecSandboxRequest carries a first-class ``stdin`` bytes field, so config-file content is piped verbatim into ``cat``: the bytes never touch the argv nor the (single-line-only, size-capped) sandbox environment. """ - timeout = self._executor_config.request_timeout_seconds + timeout = timeout if timeout is not None else self._executor_config.request_timeout_seconds request = pb.ExecSandboxRequest(sandbox_id=sandbox_id, command=command, timeout_seconds=timeout) if stdin is not None: request.stdin = stdin @@ -733,6 +768,33 @@ async def _deliver_config_files( ) return None + async def _readiness_pending(self, sandbox_id: str, container: Container) -> BackendStatusUpdate | None: + """A STARTING update while the workload is not yet reachable, else None. + + Probed from inside the sandbox against loopback, so readiness does not depend on + the gateway route or its TLS. Because a port is exposed only once this passes, + the fast path's "endpoints exist -> READY" stays sticky and never re-probes, so a + momentarily refusing port cannot flap a serving deployment. A workload that never + becomes reachable stays STARTING; the reconciler's starting-timeout is the + progress deadline that eventually fails it. + """ + probe_command = _readiness_probe_command(container) + if probe_command is None: + return None + command, description, exec_timeout = probe_command + exit_code, _ = await self._exec_detached(sandbox_id, command, timeout=exec_timeout) + # Readiness fails closed (liveness fails open): the gate admits only positive + # proof of reachability, so a flaky probe never exposes an unready workload. + # - exit 0 -> reachable; expose the port and read READY + # - nonzero -> not reachable yet; stay STARTING + # - no exit event -> undecidable; stay STARTING and re-probe next poll (the port + # is not exposed while pending, so this self-heals) + # A workload that can never be probed never claims READY -- the contract this gate + # keeps. Timeouts and RPC errors surface as UNKNOWN upstream, not as a None exit. + if exit_code == 0: + return None + return BackendStatusUpdate(status="STARTING", status_message=f"Awaiting readiness: {description}") + async def _list_endpoints(self, sandbox_nm: str) -> list[Endpoint]: """Return the sandbox's currently exposed services as endpoints.""" try: @@ -756,6 +818,87 @@ async def _expose_ports(self, sandbox_nm: str, container: Container) -> list[End return endpoints +def _resolve_probe_port(port: int | str, container: Container) -> int | None: + """Resolve a probe port (a number, or a container-port name) to a number, or None.""" + if isinstance(port, int): + return port + for declared in container.ports: + if declared.name == port: + return declared.container_port + return None + + +def _default_probe_port(container: Container) -> int | None: + """The first TCP container port, used for the default reachability probe. + + UDP ports are skipped: a TCP connect against a UDP listener never succeeds and would + wedge a healthy workload in STARTING until the progress deadline. + """ + for declared in container.ports: + if declared.protocol == "TCP": + return declared.container_port + return None + + +def _loopback_probe_script(program: str, *args: str) -> str: + """Wrap a python probe *program* so it runs against the sandbox's own python. + + Selects ``python3`` then ``python`` off PATH and runs *program* with *args*, exiting + 0 when the probe connects and nonzero when it does not. If neither interpreter is on + PATH the probe cannot run, so it exits 0 to preserve the prior expose-on-alive + behaviour rather than wedging a workload in STARTING until the progress deadline. + """ + quoted_args = " ".join(shlex.quote(arg) for arg in args) + return ( + "if command -v python3 >/dev/null 2>&1; then _py=python3; " + "elif command -v python >/dev/null 2>&1; then _py=python; " + "else exit 0; fi; " + f'"$_py" -c {shlex.quote(program)} {quoted_args}' + ) + + +def _readiness_probe_command(container: Container) -> tuple[list[str], str, int] | None: + """The in-sandbox probe: (command, description, exec_timeout_seconds), or None. + + Succeeds (exit 0) once the workload is reachable. Honours a declared readinessProbe + (exec/httpGet/tcpSocket); with no probe declared, falls back to a TCP connect on the + first TCP container port. Returns None when there is nothing to probe (no declared + probe and no TCP port), meaning "treat as ready" -- a portless (or UDP-only) workload + has no TCP socket a caller could reach anyway. + + ``exec_timeout_seconds`` bounds the ExecSandbox RPC so a hung probe cannot stall the + serial reconcile loop: an exec probe is bounded by its own ``timeoutSeconds``; a + network probe self-times in python, so the RPC is given that timeout plus headroom. + """ + probe = container.readiness_probe + timeout = probe.timeout_seconds if probe is not None else _DEFAULT_READINESS_TIMEOUT_SECONDS + network_exec_timeout = timeout + _READINESS_EXEC_TIMEOUT_MARGIN_SECONDS + + if probe is not None and probe.exec_action is not None and probe.exec_action.command: + return list(probe.exec_action.command), "exec readiness probe", timeout + + # A declared probe naming a port that does not resolve falls back to the first TCP + # port rather than skipping the gate, so a misconfigured probe cannot silently + # re-open the bind race. + if probe is not None and probe.http_get is not None: + port = _resolve_probe_port(probe.http_get.port, container) or _default_probe_port(container) + if port is None: + return None + path = probe.http_get.path if probe.http_get.path.startswith("/") else f"/{probe.http_get.path}" + url = f"{probe.http_get.scheme.lower()}://127.0.0.1:{port}{path}" + script = _loopback_probe_script(_HTTP_PROBE_PROGRAM, url, str(timeout)) + return ["/bin/sh", "-c", script], f"httpGet {url}", network_exec_timeout + + if probe is not None and probe.tcp_socket is not None: + port = _resolve_probe_port(probe.tcp_socket.port, container) or _default_probe_port(container) + else: + port = _default_probe_port(container) + if port is None: + return None + script = _loopback_probe_script(_TCP_PROBE_PROGRAM, "127.0.0.1", str(port), str(timeout)) + return ["/bin/sh", "-c", script], f"tcp 127.0.0.1:{port}", network_exec_timeout + + def _sandbox_name(workspace: str, name: str) -> str: """OpenShell sandbox name, within ``_MAX_ROUTABLE_NAME_LEN``. diff --git a/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py b/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py index 510cb6d248..ea211039cc 100644 --- a/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py +++ b/plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py @@ -5,6 +5,7 @@ from __future__ import annotations +import socket from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -614,6 +615,86 @@ async def test_read_status_ready_when_running_without_probe( assert update.status == "READY" +def _running_container_with_published_port(host_port: int) -> MagicMock: + container = MagicMock() + container.id = "abc123def456" + container.status = "running" + container.labels = { + "managed-by": MANAGED_BY_LABEL, + DEPLOYMENT_WORKSPACE_LABEL: "default", + DEPLOYMENT_NAME_LABEL: "srv", + RESTART_POLICY_LABEL: "Always", + CONFIG_NAME_LABEL: "cfg1", + RESOURCE_SCOPE_LABEL: DEFAULT_RESOURCE_SCOPE, + } + container.ports = {"8000/tcp": [{"HostPort": str(host_port)}]} + container.attrs = container_attrs() + return container + + +@pytest.mark.asyncio +async def test_read_status_ready_when_running_port_bound( + docker_backend: DockerDeploymentBackend, + mock_entities: AsyncMock, + mock_docker_client: MagicMock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # No declared probe, published port accepting connections -> READY. + monkeypatch.setenv("NMP_LOOPBACK_ADDRESS", "127.0.0.1") + mock_entities.get.return_value = sample_config() + with socket.socket() as server: + server.bind(("127.0.0.1", 0)) + server.listen(1) + host_port = server.getsockname()[1] + mock_docker_client.containers.get.return_value = _running_container_with_published_port(host_port) + + update = await docker_backend.read_status(workspace="default", name="srv") + + assert update.status == "READY" + + +@pytest.mark.asyncio +async def test_read_status_starting_when_running_port_not_bound( + docker_backend: DockerDeploymentBackend, + mock_entities: AsyncMock, + mock_docker_client: MagicMock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # No declared probe, nothing yet listening on the published port -> STARTING, so + # READY does not race the workload's bind(). Hold the port bound-but-not-listening + # for the whole probe so nothing else can bind and listen on it mid-test; a + # connect() still gets ECONNREFUSED, the not-yet-bound state under test. + monkeypatch.setenv("NMP_LOOPBACK_ADDRESS", "127.0.0.1") + mock_entities.get.return_value = sample_config() + with socket.socket() as probe_socket: + probe_socket.bind(("127.0.0.1", 0)) + host_port = probe_socket.getsockname()[1] + mock_docker_client.containers.get.return_value = _running_container_with_published_port(host_port) + + update = await docker_backend.read_status(workspace="default", name="srv") + + assert update.status == "STARTING" + assert "not ready" in update.status_message + + +@pytest.mark.asyncio +async def test_read_status_ready_when_running_udp_only_port( + docker_backend: DockerDeploymentBackend, + mock_entities: AsyncMock, + mock_docker_client: MagicMock, +) -> None: + # A UDP-only workload has no TCP listener, so the default TCP probe is skipped and + # running implies ready rather than wedging STARTING until the progress deadline. + mock_entities.get.return_value = sample_config() + container = _running_container_with_published_port(0) + container.ports = {"9000/udp": [{"HostPort": "34567"}]} + mock_docker_client.containers.get.return_value = container + + update = await docker_backend.read_status(workspace="default", name="srv") + + assert update.status == "READY" + + def _running_server_container() -> MagicMock: container = MagicMock() container.id = "abc123def456" diff --git a/plugins/nemo-deployments/tests/unit/backends/docker/test_probes.py b/plugins/nemo-deployments/tests/unit/backends/docker/test_probes.py index 33b21cf002..df1f6004cd 100644 --- a/plugins/nemo-deployments/tests/unit/backends/docker/test_probes.py +++ b/plugins/nemo-deployments/tests/unit/backends/docker/test_probes.py @@ -5,6 +5,7 @@ from __future__ import annotations +import socket from unittest.mock import MagicMock import pytest @@ -40,3 +41,58 @@ async def test_tcp_probe_without_host_url_not_ready() -> None: assert ready is False assert "no host_url available" in reason + + +@pytest.mark.asyncio +async def test_no_probe_without_host_port_is_ready() -> None: + # A portless workload has no socket to connect to, so running implies ready. + ready, reason = await check_readiness_probe( + container=MagicMock(), + probe=None, + host_url=None, + host_ports={}, + ) + + assert ready is True + assert reason == "no readiness probe configured" + + +@pytest.mark.asyncio +async def test_no_probe_with_bound_host_port_is_ready() -> None: + # No declared probe but the published port accepts a connection -> ready. + with socket.socket() as server: + server.bind(("127.0.0.1", 0)) + server.listen(1) + host_port = server.getsockname()[1] + + ready, reason = await check_readiness_probe( + container=MagicMock(), + probe=None, + host_url=f"http://127.0.0.1:{host_port}", + host_ports={8000: host_port}, + ) + + assert ready is True + assert "connected" in reason + + +@pytest.mark.asyncio +async def test_no_probe_with_unbound_host_port_not_ready() -> None: + # No declared probe and nothing listening on the published port -> not ready, so + # READY does not race the workload's bind(). Hold the port bound-but-not-listening + # for the whole probe: that reserves it (a second bind gets EADDRINUSE, so nothing + # else can grab it and start listening mid-test) while a connect() still gets + # ECONNREFUSED, which is exactly the not-yet-bound state under test. + with socket.socket() as probe_socket: + probe_socket.bind(("127.0.0.1", 0)) + host_port = probe_socket.getsockname()[1] + + ready, reason = await check_readiness_probe( + container=MagicMock(), + probe=None, + host_url=f"http://127.0.0.1:{host_port}", + host_ports={8000: host_port}, + ) + + assert ready is False + assert "not ready" in reason diff --git a/plugins/nemo-deployments/tests/unit/backends/openshell/test_openshell_backend_mocked.py b/plugins/nemo-deployments/tests/unit/backends/openshell/test_backend.py similarity index 82% rename from plugins/nemo-deployments/tests/unit/backends/openshell/test_openshell_backend_mocked.py rename to plugins/nemo-deployments/tests/unit/backends/openshell/test_backend.py index 73615521df..40a9542a94 100644 --- a/plugins/nemo-deployments/tests/unit/backends/openshell/test_openshell_backend_mocked.py +++ b/plugins/nemo-deployments/tests/unit/backends/openshell/test_backend.py @@ -22,21 +22,34 @@ ) from nemo_deployments_plugin.backends.openshell.backend import ( _CONFIG_DELIVERED_MARKER, + _DEFAULT_READINESS_TIMEOUT_SECONDS, _LAUNCH_MARKER, _LIVENESS_PROBE, _MAX_ROUTABLE_NAME_LEN, + _READINESS_EXEC_TIMEOUT_MARGIN_SECONDS, _SERVE_DEAD_EXIT, _SERVE_PENDING_EXIT, _SERVE_PID_GRACE_SECONDS, _SERVE_PIDFILE, OpenShellDeploymentBackend, _delivery_script, + _readiness_probe_command, _sandbox_name, _service_name, ) from nemo_deployments_plugin.backends.registry import BACKEND_CLASSES from nemo_deployments_plugin.constants import MANAGED_BY_LABEL -from nemo_deployments_plugin.entities import ConfigFile, Container, ContainerPort, DeploymentConfig, EnvVar +from nemo_deployments_plugin.entities import ( + ConfigFile, + Container, + ContainerPort, + DeploymentConfig, + EnvVar, + ExecAction, + HTTPGetAction, + Probe, + TCPSocketAction, +) from nemo_deployments_plugin.secrets import SecretResolutionError from nemo_platform import AsyncNeMoPlatform from nemo_platform_plugin.entity_client import NemoEntityNotFoundError @@ -126,6 +139,12 @@ def _stream_without_exit(stdout: str = "") -> list[MagicMock]: return events +def _config_with_readiness(probe: Probe) -> DeploymentConfig: + cfg = _config() + cfg.containers[0].readiness_probe = probe + return cfg + + def _config_with_env(env: list[EnvVar]) -> DeploymentConfig: return DeploymentConfig( name="cfg1", @@ -452,6 +471,104 @@ async def test_read_status_exposes_after_launch( assert [e.url for e in update.endpoints] == ["http://nmp-x--http.openshell.localhost:17670/"] +async def test_read_status_starting_until_default_tcp_probe_passes( + openshell_backend: OpenShellDeploymentBackend, mock_stub: MagicMock, mock_entities: AsyncMock +) -> None: + # No declared probe: the port must accept a connection before we expose it. Alive but + # not yet reachable -> STARTING, and no port is exposed (exposing reads as READY). + mock_entities.get.return_value = _config() + mock_stub.GetSandbox.return_value = _sandbox(pb.SANDBOX_PHASE_READY) + mock_stub.ExecSandbox.side_effect = [ + _exec_events(0), # marker present + _exec_events(0), # liveness: alive + _exec_events(1), # readiness: port not yet accepting connections + ] + + update = await openshell_backend.read_status(workspace="default", name="srv") + + assert update.status == "STARTING" + assert "readiness" in update.status_message.lower() + mock_stub.ExposeService.assert_not_called() + + +async def test_read_status_starting_when_readiness_probe_yields_no_exit( + openshell_backend: OpenShellDeploymentBackend, mock_stub: MagicMock, mock_entities: AsyncMock +) -> None: + # Readiness fails closed: a probe whose exec stream ends without an exit event cannot + # prove reachability, so it must not expose the port and flip to sticky READY. It + # stays STARTING and self-heals on the next poll -- driven here by polling read_status + # twice, undecidable then reachable, and asserting the second poll exposes and reads + # READY. (Liveness fails open on the same undecidable signal; readiness gates + # admission the other way.) + mock_entities.get.return_value = _config() + mock_stub.GetSandbox.return_value = _sandbox(pb.SANDBOX_PHASE_READY) + stdout_only = MagicMock() + stdout_only.HasField.side_effect = lambda field: field == "stdout" + stdout_only.stdout.data = b"partial" + mock_stub.ExecSandbox.side_effect = [ + # First poll: alive but readiness stream ends with no exit event -> undecidable. + _exec_events(0), # marker present + _exec_events(0), # liveness: alive + [stdout_only], # readiness: no exit event + # Second poll: still unexposed (first poll withheld the port), now reachable. + _exec_events(0), # marker present + _exec_events(0), # liveness: alive + _exec_events(0), # readiness: reachable + ] + mock_stub.ExposeService.return_value = MagicMock(url="http://nmp-x--http.openshell.localhost:17670/") + + first = await openshell_backend.read_status(workspace="default", name="srv") + + assert first.status == "STARTING" + assert "readiness" in first.status_message.lower() + mock_stub.ExposeService.assert_not_called() + + second = await openshell_backend.read_status(workspace="default", name="srv") + + assert second.status == "READY" + mock_stub.ExposeService.assert_called_once() + + +async def test_read_status_ready_when_default_tcp_probe_connects( + openshell_backend: OpenShellDeploymentBackend, mock_stub: MagicMock, mock_entities: AsyncMock +) -> None: + # Alive and the port now accepts a connection -> expose -> READY. + mock_entities.get.return_value = _config() + mock_stub.GetSandbox.return_value = _sandbox(pb.SANDBOX_PHASE_READY) + mock_stub.ExecSandbox.side_effect = [ + _exec_events(0), # marker present + _exec_events(0), # liveness: alive + _exec_events(0), # readiness: reachable + ] + mock_stub.ExposeService.return_value = MagicMock(url="http://nmp-x--http.openshell.localhost:17670/") + + update = await openshell_backend.read_status(workspace="default", name="srv") + + assert update.status == "READY" + assert [e.url for e in update.endpoints] == ["http://nmp-x--http.openshell.localhost:17670/"] + + +async def test_read_status_gates_on_declared_httpget_probe( + openshell_backend: OpenShellDeploymentBackend, mock_stub: MagicMock, mock_entities: AsyncMock +) -> None: + # A declared httpGet readinessProbe is honoured against loopback; a failing probe + # keeps the deployment STARTING and unexposed, and the probe hits the declared path. + mock_entities.get.return_value = _config_with_readiness(Probe(httpGet=HTTPGetAction(path="/health", port=8000))) + mock_stub.GetSandbox.return_value = _sandbox(pb.SANDBOX_PHASE_READY) + mock_stub.ExecSandbox.side_effect = [ + _exec_events(0), # marker present + _exec_events(0), # liveness: alive + _exec_events(1), # readiness: /health not answering yet + ] + + update = await openshell_backend.read_status(workspace="default", name="srv") + + assert update.status == "STARTING" + mock_stub.ExposeService.assert_not_called() + probe_scripts = [" ".join(c.args[0].command) for c in mock_stub.ExecSandbox.call_args_list] + assert any("http://127.0.0.1:8000/health" in script for script in probe_scripts) + + async def test_read_status_fails_when_serve_process_died( openshell_backend: OpenShellDeploymentBackend, mock_stub: MagicMock ) -> None: @@ -606,6 +723,86 @@ async def test_read_status_cleanup_swallows_delete_error( mock_stub.DeleteSandbox.assert_called_once() +def _require_probe_command(container: Container) -> tuple[list[str], str, int]: + """Return the container's readiness probe command, asserting it exists (narrows None).""" + probe_command = _readiness_probe_command(container) + assert probe_command is not None + return probe_command + + +def test_readiness_probe_command_defaults_to_tcp_on_the_first_port() -> None: + command, description, timeout = _require_probe_command(_config().containers[0]) + assert command[:2] == ["/bin/sh", "-c"] + assert "127.0.0.1" in command[2] + assert description == "tcp 127.0.0.1:8000" + # A network probe self-times in python; the RPC gets that timeout plus headroom. + assert timeout == _DEFAULT_READINESS_TIMEOUT_SECONDS + _READINESS_EXEC_TIMEOUT_MARGIN_SECONDS + + +def test_readiness_probe_command_uses_declared_httpget() -> None: + container = _config_with_readiness(Probe(httpGet=HTTPGetAction(path="/ready", port=8000))).containers[0] + command, description, _timeout = _require_probe_command(container) + assert "http://127.0.0.1:8000/ready" in command[2] + assert description == "httpGet http://127.0.0.1:8000/ready" + + +def test_readiness_probe_command_runs_declared_exec_directly() -> None: + probe = Probe(exec=ExecAction(command=["/bin/true"]), timeoutSeconds=4) + container = _config_with_readiness(probe).containers[0] + command, description, timeout = _require_probe_command(container) + assert command == ["/bin/true"] + assert description == "exec readiness probe" + # An exec probe is bounded by its own timeoutSeconds, not the control-plane deadline, + # so a hung probe cannot stall the serial reconcile loop. + assert timeout == 4 + + +def test_readiness_probe_command_resolves_named_tcp_socket_port() -> None: + container = _config_with_readiness(Probe(tcpSocket=TCPSocketAction(port="http"))).containers[0] + _command, description, _timeout = _require_probe_command(container) + assert description == "tcp 127.0.0.1:8000" + + +def test_readiness_probe_command_is_none_without_probe_or_ports() -> None: + assert _readiness_probe_command(_config(with_port=False).containers[0]) is None + + +def test_readiness_probe_command_https_uses_unverified_context() -> None: + container = _config_with_readiness( + Probe(httpGet=HTTPGetAction(path="/health", port=8000, scheme="HTTPS")) + ).containers[0] + command, _description, _timeout = _require_probe_command(container) + assert "https://127.0.0.1:8000/health" in command[2] + # An https probe against a loopback/self-signed cert must not fail verification. + assert "_create_unverified_context" in command[2] + + +def test_readiness_probe_command_normalizes_httpget_path_without_leading_slash() -> None: + container = _config_with_readiness(Probe(httpGet=HTTPGetAction(path="ready", port=8000))).containers[0] + _command, description, _timeout = _require_probe_command(container) + assert description == "httpGet http://127.0.0.1:8000/ready" + + +def test_readiness_probe_command_skips_udp_only_ports() -> None: + container = Container( + name="web", + image="img:latest", + command=["serve"], + ports=[ContainerPort(containerPort=9000, name="udp", protocol="UDP")], + ) + assert _readiness_probe_command(container) is None + + +def test_readiness_probe_command_falls_back_when_declared_port_unresolvable() -> None: + # A declared probe naming a port absent from the container falls back to the first + # TCP port rather than skipping the gate (which would re-open the bind race). + container = _config_with_readiness(Probe(httpGet=HTTPGetAction(path="/health", port="does-not-exist"))).containers[ + 0 + ] + _command, description, _timeout = _require_probe_command(container) + assert description == "httpGet http://127.0.0.1:8000/health" + + def _run_probe(tmp_path: Path, *, pid: str | None, marker: str | None) -> int: """Run the real liveness probe against tmp files, returning its exit code.