From 308403a880336abc72739ec9caae0f23a471c542 Mon Sep 17 00:00:00 2001 From: Jorge Garcia Date: Wed, 16 Sep 2026 21:59:55 -0700 Subject: [PATCH 1/3] fix(sandbox): stream workspaces to Kubernetes pods --- clearwing/sandbox/hunter_sandbox.py | 7 +++++++ tests/test_kubernetes_sandbox.py | 19 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/clearwing/sandbox/hunter_sandbox.py b/clearwing/sandbox/hunter_sandbox.py index c0e87f2f..0e910108 100644 --- a/clearwing/sandbox/hunter_sandbox.py +++ b/clearwing/sandbox/hunter_sandbox.py @@ -346,6 +346,13 @@ def spawn( Returns a SandboxInstance ready for exec/write/read. """ + if self.backend.name == "kubernetes": + # Kubernetes sandboxes do not share the caller's filesystem. + # Stream the repository after pod startup and use the image's + # pod-local /scratch directory instead of requesting host mounts. + writable_workspace = True + scratch_mount = False + if self._environment_ref is None: self.prepare_environment() diff --git a/tests/test_kubernetes_sandbox.py b/tests/test_kubernetes_sandbox.py index 560fdf0a..af33c6e1 100644 --- a/tests/test_kubernetes_sandbox.py +++ b/tests/test_kubernetes_sandbox.py @@ -42,6 +42,25 @@ def test_host_mounts_are_rejected(): backend.create("sandbox:1", config) +def test_hunter_sandbox_streams_workspace_and_uses_pod_local_scratch(tmp_path): + from clearwing.sandbox.hunter_sandbox import HunterSandbox + + backend = MagicMock(name="backend") + backend.name = "kubernetes" + backend.available_cpus.return_value = (None, "Kubernetes scheduler") + backend.ensure_environment.return_value = SimpleNamespace(reference="sandbox:1", cached=True) + instance = MagicMock() + instance.scratch_host_dir = None + backend.create.return_value = instance + manager = HunterSandbox(str(tmp_path), backend=backend) + + manager.spawn() + + config = backend.create.call_args.args[1] + assert config.mounts == [] + instance.copy_tree_into.assert_called_once_with(str(tmp_path), "/workspace") + + def test_pod_lifecycle(monkeypatch): client = pytest.importorskip("kubernetes.client") api = MagicMock() From 3fb45ab5b20a48c89ce652120c9fe121295d2939 Mon Sep 17 00:00:00 2001 From: Jorge Garcia Date: Wed, 16 Sep 2026 22:17:50 -0700 Subject: [PATCH 2/3] fix(sandbox): stream Kubernetes stdin without half-close --- clearwing/sandbox/kubernetes.py | 10 +++++++++- tests/test_kubernetes_sandbox.py | 20 ++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/clearwing/sandbox/kubernetes.py b/clearwing/sandbox/kubernetes.py index 3bf0ae96..38a5c4b1 100644 --- a/clearwing/sandbox/kubernetes.py +++ b/clearwing/sandbox/kubernetes.py @@ -233,6 +233,15 @@ def _stream(self, command: list[str], *, stdin: bytes | None = None): raise RuntimeError("sandbox pod has not been started") from kubernetes.stream import stream + if stdin is not None: + # The Kubernetes Python WSClient has no half-close operation for + # stdin. Make the remote process consume an exact byte count so it + # can finish without waiting for an EOF that the client cannot send. + command = [ + "/bin/sh", + "-c", + f"head -c {len(stdin)} | {shlex.join(command)}", + ] response = stream( self._api_factory().connect_get_namespaced_pod_exec, self._pod_name, @@ -248,7 +257,6 @@ def _stream(self, command: list[str], *, stdin: bytes | None = None): if stdin is not None: for offset in range(0, len(stdin), 1024 * 1024): response.write_stdin(stdin[offset : offset + 1024 * 1024]) - response.close_stdin() return response def exec( diff --git a/tests/test_kubernetes_sandbox.py b/tests/test_kubernetes_sandbox.py index af33c6e1..50a75dbc 100644 --- a/tests/test_kubernetes_sandbox.py +++ b/tests/test_kubernetes_sandbox.py @@ -88,3 +88,23 @@ def test_exit_code_parsing(): '{"status":"Failure","details":{"causes":[{"reason":"ExitCode","message":"137"}]}}' ) assert KubernetesSandbox._parse_exit_code(response) == 137 + + +def test_stream_with_stdin_uses_exact_length_instead_of_half_close(monkeypatch): + import kubernetes.stream + + response = MagicMock(spec=["write_stdin"]) + stream = MagicMock(return_value=response) + monkeypatch.setattr(kubernetes.stream, "stream", stream) + sandbox = KubernetesSandbox("sandbox:1", SandboxRunConfig(), namespace="jobs") + sandbox._pod_name = "clearwing-sandbox-test" + + result = sandbox._stream(["tar", "-xf", "-", "-C", "/workspace"], stdin=b"payload") + + assert result is response + assert stream.call_args.kwargs["command"] == [ + "/bin/sh", + "-c", + "head -c 7 | tar -xf - -C /workspace", + ] + response.write_stdin.assert_called_once_with(b"payload") From afd79eaebc7060ab05af5abc21c3fa4855876fbe Mon Sep 17 00:00:00 2001 From: Jorge Garcia Date: Wed, 16 Sep 2026 22:56:10 -0700 Subject: [PATCH 3/3] fix(sandbox): harden Kubernetes pod cleanup --- clearwing/sandbox/hunter_sandbox.py | 52 +++++++++++---------- clearwing/sandbox/kubernetes.py | 57 +++++++++++++++++++---- docs/sandbox-backends.md | 14 ++++++ tests/test_kubernetes_sandbox.py | 70 +++++++++++++++++++++++++++++ 4 files changed, 162 insertions(+), 31 deletions(-) diff --git a/clearwing/sandbox/hunter_sandbox.py b/clearwing/sandbox/hunter_sandbox.py index 0e910108..5997abb4 100644 --- a/clearwing/sandbox/hunter_sandbox.py +++ b/clearwing/sandbox/hunter_sandbox.py @@ -400,30 +400,36 @@ def spawn( sb = self.backend.create(environment_ref, cfg) _t = time.monotonic() - sb.start() - logger.debug( - "Sandbox instance started environment=%s in %.2fs", - environment_ref, - time.monotonic() - _t, - ) + try: + sb.start() + logger.debug( + "Sandbox instance started environment=%s in %.2fs", + environment_ref, + time.monotonic() - _t, + ) - if writable_workspace: - sb.copy_tree_into(self.repo_path, "/workspace") - try: - baseline = sb.exec( - "cd /workspace && " - "rm -rf /workspace/.git && " - "find . -name .git -type f -delete && " - "git init -q && git add -A && " - "git -c user.name=clearwing -c user.email=clearwing@localhost " - "commit -m initial -q && git rev-parse HEAD", - timeout=120, - ) - if baseline.exit_code != 0: - raise RuntimeError(baseline.stderr or "git baseline creation failed") - sb.workspace_baseline_commit = baseline.stdout.strip() - except Exception: - logger.warning("git init in writable workspace failed", exc_info=True) + if writable_workspace: + sb.copy_tree_into(self.repo_path, "/workspace") + try: + baseline = sb.exec( + "cd /workspace && " + "rm -rf /workspace/.git && " + "find . -name .git -type f -delete && " + "git init -q && git add -A && " + "git -c user.name=clearwing -c user.email=clearwing@localhost " + "commit -m initial -q && git rev-parse HEAD", + timeout=120, + ) + if baseline.exit_code != 0: + raise RuntimeError(baseline.stderr or "git baseline creation failed") + sb.workspace_baseline_commit = baseline.stdout.strip() + except Exception: + logger.warning("git init in writable workspace failed", exc_info=True) + except BaseException: + sb.stop() + if scratch_host_dir: + shutil.rmtree(scratch_host_dir, ignore_errors=True) + raise # Stash scratch host dir + variant on the container for cleanup / introspection sb.scratch_host_dir = scratch_host_dir diff --git a/clearwing/sandbox/kubernetes.py b/clearwing/sandbox/kubernetes.py index 38a5c4b1..3cd7e5a5 100644 --- a/clearwing/sandbox/kubernetes.py +++ b/clearwing/sandbox/kubernetes.py @@ -9,6 +9,7 @@ import io import json +import logging import os import shlex import subprocess @@ -26,7 +27,8 @@ from .container import ExecResult _STARTUP_TIMEOUT_SECONDS = 120 -_DEADLINE_FLOOR_SECONDS = 900 + +logger = logging.getLogger(__name__) def kubernetes_namespace() -> str: @@ -174,20 +176,20 @@ def start(self) -> str: allow_privilege_escalation=False, capabilities=client.V1Capabilities(drop=["ALL"], add=["SYS_PTRACE"]), ) + api = self._api_factory() + owner_references = self._owner_references(api=api, client=client) pod = client.V1Pod( metadata=client.V1ObjectMeta( name=pod_name, namespace=self.namespace, labels={"app.kubernetes.io/managed-by": "clearwing", "purpose": "sandbox"}, + owner_references=owner_references, ), spec=client.V1PodSpec( automount_service_account_token=False, enable_service_links=False, host_network=False, restart_policy="Never", - active_deadline_seconds=max( - _DEADLINE_FLOOR_SECONDS, self._config.timeout_seconds * 3 - ), containers=[ client.V1Container( name="sandbox", @@ -206,9 +208,11 @@ def start(self) -> str: ], ), ) - api = self._api_factory() api.create_namespaced_pod(self.namespace, pod) self._pod_name = pod_name + from .registry import ContainerRegistry + + ContainerRegistry.get().register(self) try: self._wait_until_running(api) except BaseException: @@ -216,6 +220,30 @@ def start(self) -> str: raise return pod_name + def _owner_references(self, *, api: Any, client: Any) -> list[Any] | None: + """Make Kubernetes garbage-collect a sandbox if its caller pod disappears.""" + caller_name = os.environ.get("HOSTNAME", "").strip() + if not caller_name: + return None + try: + caller = api.read_namespaced_pod(caller_name, self.namespace) + uid = caller.metadata.uid + if not isinstance(uid, str) or not uid: + return None + return [ + client.V1OwnerReference( + api_version="v1", + kind="Pod", + name=caller_name, + uid=uid, + block_owner_deletion=False, + controller=False, + ) + ] + except Exception: + logger.debug("Could not attach sandbox to caller pod", exc_info=True) + return None + def _wait_until_running(self, api: Any) -> None: deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS while time.monotonic() < deadline: @@ -348,16 +376,29 @@ def copy_tree_into(self, host_path: str, container_path: str = "/workspace") -> def stop(self) -> None: if self._pod_name is None: return - pod_name, self._pod_name = self._pod_name, None + pod_name = self._pod_name try: self._api_factory().delete_namespaced_pod( pod_name, self.namespace, grace_period_seconds=5 ) - except Exception: - pass + except Exception as error: + if getattr(error, "status", None) != 404: + logger.warning("Failed to delete Kubernetes sandbox pod %s", pod_name, exc_info=True) + return + from .registry import ContainerRegistry + + ContainerRegistry.get().unregister(self) + self._pod_name = None def __enter__(self) -> KubernetesSandbox: return self def __exit__(self, exc_type, exc_value, traceback) -> None: self.stop() + + def __del__(self) -> None: + if self._pod_name is not None: + try: + self.stop() + except Exception: + logger.debug("Kubernetes sandbox garbage-collection cleanup failed", exc_info=True) diff --git a/docs/sandbox-backends.md b/docs/sandbox-backends.md index 1eb1f325..45a3326a 100644 --- a/docs/sandbox-backends.md +++ b/docs/sandbox-backends.md @@ -111,6 +111,20 @@ The returned `SandboxInstance` provides `start`, `exec`, `write_file`, lifecycle properties. Downstream SourceHunt code depends on this protocol, not on the Docker SDK implementation. +### Kubernetes lifecycle + +Kubernetes sandbox pods intentionally have no pod-wide +`activeDeadlineSeconds`: SourceHunts may legitimately outlive the default +per-command timeout. Individual `exec` calls still enforce their requested +timeouts, and the caller owns the overall SourceHunt deadline. + +Clearwing deletes each pod during normal cleanup, retries a failed deletion at +process exit, and records the calling pod as an owner when it can be resolved. +The owner reference is the hard-crash safety net: Kubernetes garbage-collects +the sandbox after the calling pod is deleted even when Clearwing cannot run +cleanup. Deployments should apply a TTL to completed caller Jobs so that this +fallback is timely. + ## JSON-RPC protocol The socket adapter uses JSON-RPC 2.0. Each request or response is one UTF-8 diff --git a/tests/test_kubernetes_sandbox.py b/tests/test_kubernetes_sandbox.py index 50a75dbc..f7056a77 100644 --- a/tests/test_kubernetes_sandbox.py +++ b/tests/test_kubernetes_sandbox.py @@ -61,6 +61,26 @@ def test_hunter_sandbox_streams_workspace_and_uses_pod_local_scratch(tmp_path): instance.copy_tree_into.assert_called_once_with(str(tmp_path), "/workspace") +def test_hunter_sandbox_stops_pod_when_workspace_transfer_fails(tmp_path): + from clearwing.sandbox.hunter_sandbox import HunterSandbox + + backend = MagicMock(name="backend") + backend.name = "kubernetes" + backend.available_cpus.return_value = (None, "Kubernetes scheduler") + backend.ensure_environment.return_value = SimpleNamespace(reference="sandbox:1", cached=True) + instance = MagicMock() + instance.scratch_host_dir = None + instance.copy_tree_into.side_effect = RuntimeError("transfer failed") + backend.create.return_value = instance + manager = HunterSandbox(str(tmp_path), backend=backend) + + with pytest.raises(RuntimeError, match="transfer failed"): + manager.spawn() + + instance.stop.assert_called_once_with() + assert manager._spawned == [] + + def test_pod_lifecycle(monkeypatch): client = pytest.importorskip("kubernetes.client") api = MagicMock() @@ -76,12 +96,62 @@ def test_pod_lifecycle(monkeypatch): assert identity.startswith("clearwing-sandbox-") assert sandbox.is_running api.create_namespaced_pod.assert_called_once() + pod = api.create_namespaced_pod.call_args.args[1] + assert pod["spec"].active_deadline_seconds is None sandbox.stop() api.delete_namespaced_pod.assert_called_once() assert sandbox.container_id is None +def test_pod_is_owned_by_caller_pod(monkeypatch): + client = pytest.importorskip("kubernetes.client") + api = MagicMock() + api.read_namespaced_pod.return_value = SimpleNamespace( + metadata=SimpleNamespace(uid="caller-uid") + ) + api.read_namespaced_pod_status.return_value = SimpleNamespace( + status=SimpleNamespace(phase="Running") + ) + sandbox = KubernetesSandbox( + "sandbox:1", SandboxRunConfig(), namespace="jobs", api_factory=lambda: api + ) + monkeypatch.setenv("HOSTNAME", "dagster-run-123") + monkeypatch.setattr(client, "V1Pod", MagicMock(side_effect=lambda **kw: kw)) + + sandbox.start() + + pod = api.create_namespaced_pod.call_args.args[1] + owner = pod["metadata"].owner_references[0] + assert owner.name == "dagster-run-123" + assert owner.uid == "caller-uid" + sandbox.stop() + + +def test_failed_delete_remains_registered_for_retry(monkeypatch): + pytest.importorskip("kubernetes.client") + from clearwing.sandbox.registry import ContainerRegistry + + api = MagicMock() + api.read_namespaced_pod_status.return_value = SimpleNamespace( + status=SimpleNamespace(phase="Running") + ) + sandbox = KubernetesSandbox( + "sandbox:1", SandboxRunConfig(), namespace="jobs", api_factory=lambda: api + ) + monkeypatch.delenv("HOSTNAME", raising=False) + sandbox.start() + assert ContainerRegistry.get().active_count >= 1 + api.delete_namespaced_pod.side_effect = RuntimeError("temporary API failure") + + sandbox.stop() + + assert sandbox.container_id is not None + api.delete_namespaced_pod.side_effect = None + sandbox.stop() + assert sandbox.container_id is None + + def test_exit_code_parsing(): response = MagicMock() response.read_channel.return_value = (