diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 4f4cb6e7f..31662d8ce 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -125,6 +125,9 @@ deterministic scope step before routing each candidate to a fix and/or a Regression tests related to Docker/provisioner behavior: - `tests/test_docker_sandbox_mode_detection.py` (mode detection from `config.yaml`) - `tests/test_provisioner_kubeconfig.py` (kubeconfig file/directory handling) +- `tests/test_provisioner_request_threading.py` (keeps provisioner sandbox CRUD + endpoints as sync FastAPI handlers so synchronous K8s client calls run in the + Starlette worker pool instead of on the ASGI event loop) Blocking-IO runtime gate (`tests/blocking_io/`): - Wraps every item under `tests/blocking_io/` with a strict Blockbuster diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 03dee4b0c..a08a0da4a 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -54,8 +54,16 @@ def provisioner_module(): assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module + previous_module = sys.modules.get(spec.name) + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + yield module + finally: + if previous_module is None: + sys.modules.pop(spec.name, None) + else: + sys.modules[spec.name] = previous_module # --------------------------------------------------------------------------- diff --git a/backend/tests/test_provisioner_request_threading.py b/backend/tests/test_provisioner_request_threading.py new file mode 100644 index 000000000..24ea81b3a --- /dev/null +++ b/backend/tests/test_provisioner_request_threading.py @@ -0,0 +1,159 @@ +"""Regression tests for provisioner request-path K8s IO threading.""" + +from __future__ import annotations + +import asyncio +import inspect +import threading +import time +from contextlib import contextmanager +from types import SimpleNamespace + +import httpx +import pytest +from blockbuster import BlockBuster + + +class _RecordingCoreV1: + def __init__( + self, + *, + event_loop_thread_id: int, + ready_after_service_reads: dict[str, int] | None = None, + ) -> None: + self.event_loop_thread_id = event_loop_thread_id + self.thread_ids: list[int] = [] + self.service_sandboxes: set[str] = {"sandbox-existing"} + self.ready_after_service_reads = ready_after_service_reads or {} + self.service_read_counts: dict[str, int] = {} + self.created_pods: list[str] = [] + self.created_services: list[str] = [] + + def _record_k8s_call(self) -> None: + thread_id = threading.get_ident() + self.thread_ids.append(thread_id) + time.sleep(0) + if thread_id == self.event_loop_thread_id: + raise AssertionError("Kubernetes client call ran on the ASGI event-loop thread") + try: + asyncio.get_running_loop() + except RuntimeError: + return + raise AssertionError("Kubernetes client call ran inside an asyncio event loop") + + def read_namespaced_service(self, _name: str, _namespace: str): + self._record_k8s_call() + sandbox_id = _sandbox_id_from_service_name(_name) + self.service_read_counts[sandbox_id] = self.service_read_counts.get(sandbox_id, 0) + 1 + ready_after_reads = self.ready_after_service_reads.get(sandbox_id, 1) + if sandbox_id not in self.service_sandboxes or self.service_read_counts[sandbox_id] < ready_after_reads: + return _service_without_node_port(sandbox_id) + return _service(sandbox_id) + + def read_namespaced_pod(self, _name: str, _namespace: str): + self._record_k8s_call() + return SimpleNamespace(status=SimpleNamespace(phase="Running")) + + def create_namespaced_pod(self, _namespace: str, pod) -> None: + self._record_k8s_call() + sandbox_id = pod.metadata.labels["sandbox-id"] + self.created_pods.append(sandbox_id) + + def create_namespaced_service(self, _namespace: str, service) -> None: + self._record_k8s_call() + sandbox_id = service.metadata.labels["sandbox-id"] + self.created_services.append(sandbox_id) + self.service_sandboxes.add(sandbox_id) + + def delete_namespaced_service(self, _name: str, _namespace: str) -> None: + self._record_k8s_call() + + def delete_namespaced_pod(self, _name: str, _namespace: str) -> None: + self._record_k8s_call() + + def list_namespaced_service(self, _namespace: str, *, label_selector: str): + self._record_k8s_call() + assert label_selector == "app=deer-flow-sandbox" + return SimpleNamespace(items=[_service("sandbox-listed")]) + + +def _service(sandbox_id: str): + return SimpleNamespace( + metadata=SimpleNamespace(labels={"sandbox-id": sandbox_id}), + spec=SimpleNamespace(ports=[SimpleNamespace(name="http", node_port=32123)]), + ) + + +def _service_without_node_port(sandbox_id: str): + return SimpleNamespace( + metadata=SimpleNamespace(labels={"sandbox-id": sandbox_id}), + spec=SimpleNamespace(ports=[]), + ) + + +def _sandbox_id_from_service_name(name: str) -> str: + assert name.startswith("sandbox-") + assert name.endswith("-svc") + return name[len("sandbox-") : -len("-svc")] + + +@contextmanager +def _detect_provisioner_blocking_io(provisioner_module): + detector = BlockBuster(scanned_modules=[provisioner_module.__name__]) + detector.activate() + try: + yield + finally: + detector.deactivate() + + +def test_sandbox_business_route_handlers_are_sync(provisioner_module) -> None: + """FastAPI runs sync handlers in its worker pool, away from the event loop.""" + for handler in ( + provisioner_module.create_sandbox, + provisioner_module.destroy_sandbox, + provisioner_module.get_sandbox, + provisioner_module.list_sandboxes, + ): + assert not inspect.iscoroutinefunction(handler) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("method", "path", "json_body", "expected_created_sandbox"), + [ + ("POST", "/api/sandboxes", {"sandbox_id": "sandbox-existing", "thread_id": "thread-1", "user_id": "user-1"}, None), + ("POST", "/api/sandboxes", {"sandbox_id": "sandbox-new", "thread_id": "thread-1", "user_id": "user-1"}, "sandbox-new"), + ("DELETE", "/api/sandboxes/sandbox-existing", None, None), + ("GET", "/api/sandboxes/sandbox-existing", None, None), + ("GET", "/api/sandboxes", None, None), + ], + ids=["create-existing", "create-new", "destroy", "get", "list"], +) +async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread( + method: str, + path: str, + json_body: dict[str, str] | None, + expected_created_sandbox: str | None, + monkeypatch: pytest.MonkeyPatch, + provisioner_module, +) -> None: + fake_core_v1 = _RecordingCoreV1( + event_loop_thread_id=threading.get_ident(), + ready_after_service_reads={"sandbox-new": 3}, + ) + monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1) + + with _detect_provisioner_blocking_io(provisioner_module): + transport = httpx.ASGITransport(app=provisioner_module.app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + if json_body is None: + response = await client.request(method, path) + else: + response = await client.request(method, path, json=json_body) + + assert response.status_code == 200 + assert fake_core_v1.thread_ids + if expected_created_sandbox is not None: + assert fake_core_v1.created_pods == [expected_created_sandbox] + assert fake_core_v1.created_services == [expected_created_sandbox] diff --git a/docker/provisioner/README.md b/docker/provisioner/README.md index 6144bda1d..0c9a4e3c8 100644 --- a/docker/provisioner/README.md +++ b/docker/provisioner/README.md @@ -36,6 +36,13 @@ The **Sandbox Provisioner** is a FastAPI service that dynamically manages sandbo 5. **Cleanup**: When the session ends, `DELETE /api/sandboxes/{sandbox_id}` removes both the Pod and Service. +The sandbox business endpoints are implemented as synchronous FastAPI handlers +because the Kubernetes Python client used here is synchronous. Starlette runs +sync handlers in its worker pool, keeping create/read/list/delete K8s API calls +and the NodePort polling sleep off the ASGI event-loop thread. Keep `/health` +lightweight; do not move the sandbox CRUD handlers back to `async def` unless +the K8s client path is also made async or explicitly offloaded. + ## Requirements Host machine with a running Kubernetes cluster (Docker Desktop K8s, OrbStack, minikube, kind, etc.) diff --git a/docker/provisioner/app.py b/docker/provisioner/app.py index 130438d4c..8015ba5ea 100644 --- a/docker/provisioner/app.py +++ b/docker/provisioner/app.py @@ -289,7 +289,9 @@ def _build_volumes(thread_id: str) -> list[k8s_client.V1Volume]: return [skills_vol, userdata_vol] -def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list[k8s_client.V1VolumeMount]: +def _build_volume_mounts( + thread_id: str, user_id: str = DEFAULT_USER_ID +) -> list[k8s_client.V1VolumeMount]: """Build volume mount list, using subPath for PVC user-data.""" userdata_mount = k8s_client.V1VolumeMount( name="user-data", @@ -297,7 +299,9 @@ def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list read_only=False, ) if USERDATA_PVC_NAME: - userdata_mount.sub_path = f"deer-flow/users/{user_id}/threads/{thread_id}/user-data" + userdata_mount.sub_path = ( + f"deer-flow/users/{user_id}/threads/{thread_id}/user-data" + ) return [ k8s_client.V1VolumeMount( @@ -309,7 +313,9 @@ def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list ] -def _build_pod(sandbox_id: str, thread_id: str, user_id: str = DEFAULT_USER_ID) -> k8s_client.V1Pod: +def _build_pod( + sandbox_id: str, thread_id: str, user_id: str = DEFAULT_USER_ID +) -> k8s_client.V1Pod: """Construct a Pod manifest for a single sandbox.""" return k8s_client.V1Pod( metadata=k8s_client.V1ObjectMeta( @@ -442,7 +448,7 @@ async def health(): @app.post("/api/sandboxes", response_model=SandboxResponse) -async def create_sandbox(req: CreateSandboxRequest): +def create_sandbox(req: CreateSandboxRequest): """Create a sandbox Pod + NodePort Service for *sandbox_id*. If the sandbox already exists, returns the existing information @@ -470,7 +476,9 @@ async def create_sandbox(req: CreateSandboxRequest): # ── Create Pod ─────────────────────────────────────────────────── try: - core_v1.create_namespaced_pod(K8S_NAMESPACE, _build_pod(sandbox_id, thread_id, user_id=user_id)) + core_v1.create_namespaced_pod( + K8S_NAMESPACE, _build_pod(sandbox_id, thread_id, user_id=user_id) + ) logger.info(f"Created Pod {_pod_name(sandbox_id)}") except ApiException as exc: if exc.status != 409: # 409 = AlreadyExists @@ -514,7 +522,7 @@ async def create_sandbox(req: CreateSandboxRequest): @app.delete("/api/sandboxes/{sandbox_id}") -async def destroy_sandbox(sandbox_id: str): +def destroy_sandbox(sandbox_id: str): """Destroy a sandbox Pod + Service.""" errors: list[str] = [] @@ -543,7 +551,7 @@ async def destroy_sandbox(sandbox_id: str): @app.get("/api/sandboxes/{sandbox_id}", response_model=SandboxResponse) -async def get_sandbox(sandbox_id: str): +def get_sandbox(sandbox_id: str): """Return current status and URL for a sandbox.""" node_port = _get_node_port(sandbox_id) if not node_port: @@ -557,7 +565,7 @@ async def get_sandbox(sandbox_id: str): @app.get("/api/sandboxes") -async def list_sandboxes(): +def list_sandboxes(): """List every sandbox currently managed in the namespace.""" try: services = core_v1.list_namespaced_service(