diff --git a/doc/source/ray-core/sandboxes.md b/doc/source/ray-core/sandboxes.md index 8c8200c0a94e..d3f1c8ca00f9 100644 --- a/doc/source/ray-core/sandboxes.md +++ b/doc/source/ray-core/sandboxes.md @@ -323,6 +323,108 @@ Ray Sandboxes implement multi-layered defense-in-depth isolation: * **Network containment**: By default, `network="none"` disables all outbound network interfaces, which prevents untrusted code from making external API calls or scanning the internal cluster network. When internet access is needed, `network="public"` grants egress without handing over the host's resolver configuration or network identity; see [Networking and DNS](#networking-and-dns). * **Resource quotas**: cgroups enforce CPU quotas and memory limits, which prevents CPU starvation and out-of-memory (OOM) conditions from affecting other Ray actors. +## HTTP API service + +Ray Sandbox ships an experimental REST API service so you can manage sandboxes from outside the Ray cluster with nothing but an HTTP client and a bearer token. The service is a FastAPI app on Ray Serve (`ray.experimental.sandbox.http`). Each sandbox is held by a named, detached actor, so the service itself is stateless and its replicas can scale or restart without losing sandboxes. + +Image pulls and commands can far outlive an HTTP request and the load balancer in front of a deployed service, so creation and execution are asynchronous. `POST` returns immediately and clients poll, optionally long-polling with `wait_seconds` for up to 30 seconds per request. + +### Endpoints + +All endpoints sit under `/api/v1`. Except for `GET /health`, they require `Authorization: Bearer ` when a token is configured. + +| Method and path | Description | +| --- | --- | +| `GET /health` | Liveness probe; never requires auth. | +| `POST /sandboxes` | Create a sandbox. Returns `202` with `status: pending`. Poll until `running` or `error`. Send a `client_token` to make creation idempotent, so a retry returns `200` with the existing sandbox. | +| `GET /sandboxes?label=k=v` | List sandboxes, optionally filtered by labels. | +| `GET /sandboxes/{id}?wait_seconds=N` | Sandbox status; long-polls while it boots. | +| `DELETE /sandboxes/{id}` | Terminate the sandbox and its actor. Idempotent from any state. | +| `POST /sandboxes/{id}/execs` | Start a command. Returns `202` with an `exec_id`, or `409` while the sandbox isn't running. A string command runs under the sandbox's shell, `/bin/bash` by default and configurable per sandbox and per exec via `shell`. A list runs argv-style. | +| `GET /sandboxes/{id}/execs/{exec_id}?wait_seconds=N` | Exec status and result: `running`, `completed` with `exit_code`, `stdout`, and `stderr`, `timeout`, or `error`. Output is capped per stream by `max_output_bytes` with a loud truncation marker. | +| `PUT /sandboxes/{id}/files?path=/abs/path` | Write the raw request body to a file in the sandbox. Returns `413` above `max_file_bytes`. Pass `append=true` to extend the file, which lets clients chunk large uploads under proxy body-size limits. | +| `GET /sandboxes/{id}/files?path=/abs/path` | Read a file from the sandbox as `application/octet-stream`. | + +Errors use a JSON envelope of the form `{"error": {"code": "...", "message": "..."}}`. The codes are `401 unauthorized`, `404 sandbox_not_found`, `404 exec_not_found`, `404 file_not_found`, `409 conflict`, `409 unschedulable`, `400 invalid_request`, `413 payload_too_large`, and FastAPI's native `422` for schema violations. The full OpenAPI schema is served at `/openapi.json`. + +Server behavior worth knowing: + +* **TTL**: Every sandbox gets a TTL that reclaims both the sandbox and its hosting actor. Request it with `ttl_seconds`, capped and defaulted by the server's `max_ttl_seconds`. +* **Resources**: `resources` separates cluster reservations from in-sandbox cgroup caps. `cpu_request`, `memory_request_mb`, and `custom` Ray resources reserve cluster capacity, and custom resources such as `{"gvisor": 1}` pin sandboxes to runsc-equipped nodes. `cpu_limit` and `memory_limit_mb` become cgroup caps. Requests default to the limits. +* **Capabilities**: By default sandboxes get Docker's default Linux capability set so images behave the way they do under Docker. Ray's own default is far narrower and breaks `apt-get` and `tar`. The sets are written exactly, so `capabilities: []` runs the sandbox with no capabilities at all. +* **Network modes**: The modes are the Python API's, validated by Ray: `none` (the default), `public` for egress with generated DNS that `dns` overrides, `host`, and `sandbox`. See [Networking and DNS](#networking-and-dns). + +### Self-hosted quickstart + +On a Linux machine or cluster with `runsc` on `PATH`: + +```bash +pip install "ray[serve]" +export RAY_SANDBOX_API_TOKEN=dev-token # Optional. Unset disables app-level auth. +serve run ray.experimental.sandbox.http.app:build_app +``` + +```bash +curl -s -H "Authorization: Bearer dev-token" \ + -H "Content-Type: application/json" \ + -d '{"image": "busybox:latest", "readonly": false, "shell": "/bin/sh"}' \ + http://localhost:8000/api/v1/sandboxes +``` + +Builder arguments configure the server. See `ray.experimental.sandbox.http.schemas.SandboxAPISettings` for the full list. For example: + +```bash +serve run ray.experimental.sandbox.http.app:build_app max_ttl_seconds=86400 num_replicas=2 +``` + +### Deploying as an Anyscale service + +Build a cluster image whose worker nodes have `runsc`: + +```dockerfile +FROM anyscale/ray:2.58.0-py312 +RUN ARCH=$(uname -m | sed 's/arm64/aarch64/') && \ + curl -fsSL -o /usr/local/bin/runsc \ + "https://storage.googleapis.com/gvisor/releases/release/latest/${ARCH}/runsc" && \ + chmod +x /usr/local/bin/runsc +``` + +Then deploy the builder as the service's application: + +```yaml +# service.yaml +name: ray-sandbox-api +image_uri: /ray-sandbox-api:latest +applications: + - name: sandbox-api + import_path: ray.experimental.sandbox.http.app:build_app + args: + max_ttl_seconds: 86400 +``` + +```bash +anyscale service deploy -f service.yaml +``` + +Anyscale services require their own bearer token at the platform edge, so leave `RAY_SANDBOX_API_TOKEN` unset and hand clients the service's base URL and token. Consumers such as the [Harbor](https://harborframework.com) `ray-sandbox` environment take exactly that pair as `RAY_SANDBOX_API_URL` and `RAY_SANDBOX_API_KEY`. + +### Local development loop on macOS + +`runsc` is Linux-only. Develop against the service in a privileged container: + +```bash +docker run --privileged -p 8000:8000 \ + -v ~/path/to/ray/python/ray/experimental/sandbox:/overlay:ro \ + rayproject/ray:nightly-py312 bash -lc ' + pip install "ray[serve]" && + SITE=$(python -c "import ray, os; print(os.path.dirname(ray.__file__))") && + cp -r /overlay/* "$SITE/experimental/sandbox/" && + ARCH=$(uname -m | sed "s/arm64/aarch64/") && + curl -fsSL -o /usr/local/bin/runsc "https://storage.googleapis.com/gvisor/releases/release/latest/${ARCH}/runsc" && + chmod +x /usr/local/bin/runsc && + RAY_SANDBOX_API_TOKEN=dev-token serve run --host 0.0.0.0 ray.experimental.sandbox.http.app:build_app' +``` + ## API reference For detailed signatures, parameters, and return types, see {ref}`ray-sandbox-ref`. diff --git a/python/ray/experimental/sandbox/backend/base.py b/python/ray/experimental/sandbox/backend/base.py index 9eb1e52020fc..ff66fe363f3c 100644 --- a/python/ray/experimental/sandbox/backend/base.py +++ b/python/ray/experimental/sandbox/backend/base.py @@ -107,7 +107,11 @@ def exec_command( @abstractmethod def write_file( - self, sandbox_id: str, path: str, content: Union[str, bytes] + self, + sandbox_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: """Write content to a file inside the sandbox. @@ -115,6 +119,7 @@ def write_file( sandbox_id: Unique string identifier of the sandbox. path: Target file path inside the sandbox environment. content: Text string or raw bytes to write. + append: Append to the file instead of truncating it. """ pass diff --git a/python/ray/experimental/sandbox/backend/gvisor.py b/python/ray/experimental/sandbox/backend/gvisor.py index 442808799c32..5bacc902a404 100644 --- a/python/ray/experimental/sandbox/backend/gvisor.py +++ b/python/ray/experimental/sandbox/backend/gvisor.py @@ -5,6 +5,7 @@ import subprocess import time import uuid +from pathlib import Path from typing import Callable, Dict, List, Optional, Union from ray.experimental.sandbox.backend.base import ( @@ -215,6 +216,42 @@ def delete_sandbox(self, sandbox_id: str) -> None: shutil.rmtree(root_dir, ignore_errors=True) + def _resolve_exec_user(self, user: str, image: str) -> str: + """Turn a user name or uid[:gid] into runsc exec's numeric form. + + runsc only accepts numeric ids; names are resolved against the + image's own /etc/passwd (and the login group via /etc/group). + + Args: + user: Numeric uid, "uid:gid", or a user name from the image. + image: The sandbox's image, locating the extracted rootfs. + + Returns: + A "uid" or "uid:gid" string runsc accepts. + + Raises: + SandboxExecError: When a named user is not in the image's + /etc/passwd. + """ + head = user.split(":", 1)[0] + if head.isdigit(): + return user + rootfs = os.path.join(self._image_manager.get_image_dir(image), "rootfs") + try: + passwd = Path(os.path.join(rootfs, "etc", "passwd")).read_text( + encoding="utf-8", errors="replace" + ) + except OSError: + passwd = "" + for line in passwd.splitlines(): + parts = line.split(":") + if len(parts) >= 4 and parts[0] == user: + return f"{parts[2]}:{parts[3]}" + raise SandboxExecError( + f"user {user!r} not found in the image's /etc/passwd; " + "pass a numeric uid or uid:gid instead" + ) + def exec_command( self, sandbox_id: str, @@ -223,6 +260,7 @@ def exec_command( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a process inside the running gVisor sandbox instance via runsc exec.""" meta = self._get_metadata_or_raise(sandbox_id) @@ -237,6 +275,8 @@ def exec_command( # Production execution against running container via `runsc exec` runsc_args = self._runsc_base_args(config) runsc_args.extend(["exec", "-cwd", exec_cwd]) + if user is not None: + runsc_args.extend(["-user", self._resolve_exec_user(user, config.image)]) if env: for k, v in env.items(): runsc_args.extend(["-env", f"{k}={v}"]) @@ -278,9 +318,13 @@ def exec_command( raise SandboxExecError(f"gVisor exec failed: {err}") from err def write_file( - self, sandbox_id: str, path: str, content: Union[str, bytes] + self, + sandbox_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: - """Write content to a file inside the local gVisor sandbox directory.""" + """Write (or append) content to a file inside the sandbox.""" meta = self._get_metadata_or_raise(sandbox_id) config: SandboxConfig = meta["config"] @@ -294,7 +338,9 @@ def write_file( sandbox_id, "/bin/sh", "-c", - 'mkdir -p "$(dirname "$1")" && cat > "$1"', + 'mkdir -p "$(dirname "$1")" && cat >> "$1"' + if append + else 'mkdir -p "$(dirname "$1")" && cat > "$1"', "--", path, ] diff --git a/python/ray/experimental/sandbox/http/BUILD.bazel b/python/ray/experimental/sandbox/http/BUILD.bazel new file mode 100644 index 000000000000..467628825c15 --- /dev/null +++ b/python/ray/experimental/sandbox/http/BUILD.bazel @@ -0,0 +1,44 @@ +load("@rules_python//python:defs.bzl", "py_library") +load("//bazel:python.bzl", "py_test_module_list") + +py_library( + name = "sandbox_http_lib", + srcs = glob( + ["**/*.py"], + exclude = ["tests/**/*.py"], + ), + visibility = [ + "//python/ray/experimental/sandbox/http:__pkg__", + "//python/ray/experimental/sandbox/http:__subpackages__", + ], + deps = [ + "//python/ray/experimental/sandbox:sandbox_lib", + ], +) + +py_library( + name = "conftest", + srcs = ["tests/conftest.py"], + deps = [ + "//python/ray/tests:conftest", + ], +) + +py_test_module_list( + size = "medium", + files = glob( + ["tests/**/*.py"], + exclude = [ + "tests/conftest.py", + "tests/__init__.py", + ], + ), + tags = [ + "sandbox_tests", + "team:core", + ], + deps = [ + ":conftest", + ":sandbox_http_lib", + ], +) diff --git a/python/ray/experimental/sandbox/http/__init__.py b/python/ray/experimental/sandbox/http/__init__.py new file mode 100644 index 000000000000..87ee7fba319d --- /dev/null +++ b/python/ray/experimental/sandbox/http/__init__.py @@ -0,0 +1,35 @@ +"""HTTP API service for Ray Sandbox. + +This subpackage exposes ``ray.experimental.sandbox`` over a versioned REST +API (``/api/v1``) served by Ray Serve, so sandboxes can be managed from +outside the Ray cluster with nothing but an HTTP client and a bearer token. + +Importing this package requires the Serve extra +(``pip install "ray[serve]"``); the base ``ray.experimental.sandbox`` +package deliberately never imports it. +""" + +from ray.experimental.sandbox.http.app import build_app, create_app +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import ( + DOCKER_DEFAULT_CAPABILITIES, + CreateSandboxRequest, + ExecInfo, + ResourceSpec, + SandboxAPISettings, + SandboxInfo, + StartExecRequest, +) + +__all__ = [ + "build_app", + "create_app", + "SandboxHost", + "DOCKER_DEFAULT_CAPABILITIES", + "CreateSandboxRequest", + "ExecInfo", + "ResourceSpec", + "SandboxAPISettings", + "SandboxInfo", + "StartExecRequest", +] diff --git a/python/ray/experimental/sandbox/http/app.py b/python/ray/experimental/sandbox/http/app.py new file mode 100644 index 000000000000..47b07bef169f --- /dev/null +++ b/python/ray/experimental/sandbox/http/app.py @@ -0,0 +1,622 @@ +"""FastAPI application and Ray Serve builder for the Ray Sandbox HTTP API. + +Deploy on a Ray cluster whose worker nodes have gVisor's ``runsc`` on PATH: + + serve run ray.experimental.sandbox.http.app:build_app + +or as an Anyscale service (see ``doc/source/ray-core/sandboxes.md``). Bearer +auth is enforced when the environment variable named by +``SandboxAPISettings.token_env_var`` (default ``RAY_SANDBOX_API_TOKEN``) is +set; an Anyscale service can leave it unset because the platform edge already +requires the service's own bearer token. + +The service holds no state: every sandbox lives in a named detached +``SandboxHost`` actor and every replica resolves them by name, so replicas +can scale or restart freely. +""" + +import asyncio +import hashlib +import hmac +import logging +import os +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Awaitable, Dict, List, Optional + +from fastapi import APIRouter, Depends, FastAPI, Query, Request, Response +from fastapi.responses import JSONResponse + +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import ( + CreateSandboxRequest, + ExecInfo, + ExecStarted, + SandboxAPISettings, + SandboxInfo, + SandboxList, + StartExecRequest, +) +from ray.util.annotations import PublicAPI + +logger = logging.getLogger(__name__) + +SANDBOX_ID_PREFIX = "sb-" + +_MAX_WAIT_SECONDS = 30.0 + +_WAIT_QUERY = Query( + default=0.0, + ge=0.0, + le=_MAX_WAIT_SECONDS, + description="Long-poll for up to this many seconds for a state change.", +) + + +class _ApiError(Exception): + """Maps to the JSON error envelope; raised by handlers, caught app-wide.""" + + def __init__(self, status_code: int, code: str, message: str) -> None: + super().__init__(message) + self.status_code = status_code + self.code = code + self.message = message + + +def _sandbox_not_found(sandbox_id: str) -> _ApiError: + return _ApiError(404, "sandbox_not_found", f"no sandbox with id {sandbox_id!r}") + + +class _SchedulingTimeout(_ApiError): + """The sandbox's hosting actor has not been scheduled within the grace.""" + + def __init__(self, sandbox_id: str) -> None: + super().__init__( + 409, + "conflict", + f"sandbox {sandbox_id} is not ready yet (its actor is still " + "being scheduled); retry shortly", + ) + + +def _is_unschedulable(exc: BaseException) -> bool: + """True when Ray reports the actor can never fit the cluster's resources. + + Matched by class name for the same reason as ``_is_actor_gone``. + """ + names = {type(exc).__name__, *(base.__name__ for base in type(exc).__mro__)} + return any("ActorUnschedulableError" in name for name in names) + + +def _is_actor_unavailable(exc: BaseException) -> bool: + """True when a remote call failed because the actor is *transiently* + unreachable (a restart or network blip), not because it is gone. + + Ray makes ``ActorUnavailableError`` a subclass of ``RayActorError``, so + this must be checked before ``_is_actor_gone`` — otherwise a recoverable + blip is misread as a permanent death. Matched by class name for the same + importability reason as ``_is_actor_gone``. + """ + names = {type(exc).__name__, *(base.__name__ for base in type(exc).__mro__)} + return any("ActorUnavailableError" in name for name in names) + + +def _is_actor_gone(exc: BaseException) -> bool: + """True when a remote call failed because the actor no longer exists. + + Matched by class name because Ray re-raises remote failures as + dynamically-built subclasses, and so this module stays importable (and + unit-testable) without a live Ray context. Transient unreachability + (``ActorUnavailableError``) is handled separately by + ``_is_actor_unavailable`` and must be ruled out first. + """ + names = {type(exc).__name__, *(base.__name__ for base in type(exc).__mro__)} + return any("RayActorError" in name or "ActorDiedError" in name for name in names) + + +async def _actor_call(sandbox_id: str, awaitable: Awaitable[Any]) -> Any: + """Await a SandboxHost call, mapping actor failures to HTTP errors.""" + try: + return await awaitable + except Exception as exc: + if _is_actor_unavailable(exc): + # A transient blip (actor restarting / briefly unreachable), not a + # death. 503 so the caller retries the same sandbox instead of a + # client_token retry being misread as "gone" and killing it. + raise _ApiError( + 503, + "sandbox_unavailable", + f"sandbox {sandbox_id} is temporarily unavailable; retry shortly", + ) from exc + if _is_actor_gone(exc): + raise _sandbox_not_found(sandbox_id) from exc + if _is_unschedulable(exc): + # The requested cpu/memory shape cannot fit any node the cluster + # can offer: a permanent condition, not a scheduling wait. + raise _ApiError( + 409, + "unschedulable", + f"sandbox {sandbox_id} cannot be scheduled: {str(exc)[:300]}", + ) from exc + raise + + +def _sandbox_name_for_token(client_token: str) -> str: + digest = hashlib.sha256(client_token.encode("utf-8")).hexdigest() + return f"{SANDBOX_ID_PREFIX}{digest[:12]}" + + +def _parse_label_filters(labels: List[str]) -> Dict[str, str]: + filters: Dict[str, str] = {} + for item in labels: + key, sep, value = item.partition("=") + if not sep or not key: + raise _ApiError( + 400, + "invalid_request", + f"label filter {item!r} must have the form key=value", + ) + filters[key] = value + return filters + + +@PublicAPI(stability="alpha") +class RayActorHandleResolver: + """Creates and resolves the named detached SandboxHost actors. + + The default (and only production) resolver; tests inject a fake with the + same four methods so the HTTP layer runs without a Ray cluster. + """ + + def __init__(self, settings: SandboxAPISettings) -> None: + self._settings = settings + + def create( + self, + name: str, + actor_options: Dict[str, Any], + ctor_kwargs: Dict[str, Any], + ) -> Any: + import ray + + return ( + ray.remote(SandboxHost) + .options( + name=name, + namespace=self._settings.namespace, + lifetime="detached", + # Atomic get-or-create: two replicas racing on the same + # client_token converge on one actor. boot() is idempotent. + get_if_exists=True, + **actor_options, + ) + .remote(**ctor_kwargs) + ) + + def get(self, name: str) -> Optional[Any]: + import ray + + try: + return ray.get_actor(name, namespace=self._settings.namespace) + except ValueError: + return None + + def list_names(self) -> List[str]: + from ray.util import list_named_actors + + names: List[str] = [] + for entry in list_named_actors(all_namespaces=True): + if entry.get("namespace") == self._settings.namespace and entry.get( + "name", "" + ).startswith(SANDBOX_ID_PREFIX): + names.append(entry["name"]) + return names + + def kill(self, handle: Any) -> None: + import ray + + try: + ray.kill(handle) + except Exception as exc: + logger.debug("Failed to kill sandbox actor: %s", exc) + + +@PublicAPI(stability="alpha") +def create_app( + settings: Optional[SandboxAPISettings] = None, + *, + handle_resolver: Optional[Any] = None, +) -> FastAPI: + """Build the FastAPI app. + + Args: + settings: Server settings; defaults are production-safe. + handle_resolver: Test seam; anything with the + ``RayActorHandleResolver`` method surface. Defaults to the real + Ray-backed resolver. + + Returns: + The configured FastAPI application. + """ + settings = settings or SandboxAPISettings() + resolver = handle_resolver or RayActorHandleResolver(settings) + token = os.environ.get(settings.token_env_var) or None + + async def _bounded_call( + sandbox_id: str, awaitable: Awaitable[Any], extra_wait: float = 0.0 + ) -> Any: + """Await a SandboxHost call, but never block behind actor scheduling. + + A named detached actor exists the moment it is created, but on a + cluster that is scaling up it may not be *scheduled* for a while — + and a call to an unscheduled actor blocks indefinitely. Cap the wait + at the request's own long-poll budget plus a grace period. + """ + try: + return await asyncio.wait_for( + _actor_call(sandbox_id, awaitable), + timeout=extra_wait + settings.scheduling_grace_seconds, + ) + except asyncio.TimeoutError: + raise _SchedulingTimeout(sandbox_id) + + async def _describe_or_pending( + sandbox_id: str, handle: Any, wait_seconds: float = 0.0 + ) -> Dict[str, Any]: + try: + return await _bounded_call( + sandbox_id, + handle.describe.remote(wait_seconds=wait_seconds), + extra_wait=wait_seconds, + ) + except _SchedulingTimeout: + # The actor owns the spec, so report a shape-complete pending + # info with the fields only it knows left empty. + return { + "sandbox_id": sandbox_id, + "status": "pending", + "image": "", + "created_at": datetime.now(timezone.utc).isoformat(), + "ttl_seconds": None, + "expires_at": None, + "network": "none", + "labels": {}, + "error": None, + } + + async def require_bearer_token(request: Request) -> None: + if token is None: + return + provided = request.headers.get("authorization", "") + if not hmac.compare_digest(provided, f"Bearer {token}"): + raise _ApiError(401, "unauthorized", "invalid or missing bearer token") + + public = APIRouter(prefix="/api/v1") + v1 = APIRouter(prefix="/api/v1", dependencies=[Depends(require_bearer_token)]) + + @public.get("/health") + async def health() -> Dict[str, str]: + return {"status": "ok"} + + # ------------------------------------------------------------------ + # Sandboxes + # ------------------------------------------------------------------ + + @v1.post("/sandboxes", response_model=SandboxInfo, status_code=202) + async def create_sandbox(request: CreateSandboxRequest) -> JSONResponse: + if ( + request.ttl_seconds is not None + and request.ttl_seconds > settings.max_ttl_seconds + ): + raise _ApiError( + 400, + "invalid_request", + f"ttl_seconds may not exceed {settings.max_ttl_seconds}", + ) + # No TTL still gets the server-wide bound so abandoned sandboxes are + # always reclaimed eventually. + effective_ttl = ( + request.ttl_seconds + if request.ttl_seconds is not None + else settings.max_ttl_seconds + ) + + if request.client_token is not None: + sandbox_id = _sandbox_name_for_token(request.client_token) + existing = resolver.get(sandbox_id) + if existing is not None: + try: + info = await _describe_or_pending(sandbox_id, existing) + return JSONResponse(status_code=200, content=info) + except _ApiError as exc: + if exc.code != "sandbox_not_found": + raise + # The previous actor died (node loss, OOM). Clear it and + # fall through to create a fresh sandbox under the same + # name, keeping the token idempotent across failures. + resolver.kill(existing) + else: + sandbox_id = f"{SANDBOX_ID_PREFIX}{uuid.uuid4().hex[:12]}" + + resources = request.resources + cpu_request = resources.cpu_request if resources else None + cpu_limit = resources.cpu_limit if resources else None + memory_request_mb = resources.memory_request_mb if resources else None + memory_limit_mb = resources.memory_limit_mb if resources else None + # A capped sandbox should also be scheduled onto capacity that can + # honor the cap, so requests default to the limits. + if cpu_request is None: + cpu_request = cpu_limit + if memory_request_mb is None: + memory_request_mb = memory_limit_mb + + actor_options: Dict[str, Any] = { + "num_cpus": ( + cpu_request + if cpu_request is not None + else settings.default_actor_num_cpus + ), + } + if memory_request_mb is not None: + actor_options["memory"] = memory_request_mb * 1024 * 1024 + if resources is not None and resources.custom: + actor_options["resources"] = dict(resources.custom) + + capabilities = ( + request.capabilities + if request.capabilities is not None + else list(settings.default_capabilities) + ) + spec = { + "image": request.image, + "env": request.env, + "workdir": request.workdir, + "ttl_seconds": effective_ttl, + "network": request.network, + "dns": request.dns, + "shell": request.shell, + "rootless": request.rootless, + "readonly": request.readonly, + "capabilities": capabilities, + "cpu_limit": cpu_limit, + "memory_limit_mb": memory_limit_mb, + "image_pull_timeout_seconds": request.image_pull_timeout_seconds, + "start_timeout_seconds": request.start_timeout_seconds, + "labels": request.labels, + } + host_settings = { + "max_output_bytes": settings.max_output_bytes, + "max_exec_history": settings.max_exec_history, + "auto_install_runsc": settings.auto_install_runsc, + } + handle = resolver.create( + sandbox_id, + actor_options, + { + "sandbox_id": sandbox_id, + "spec": spec, + "settings": host_settings, + }, + ) + # Fire-and-forget: boot progress and failures are reported through + # describe(), never through this call's result. The response is + # synthesized from the request rather than fetched from the actor so + # that creation never blocks behind actor scheduling — on a saturated + # cluster the new actor may legitimately be queued for a while. + handle.boot.remote() + created_at = datetime.now(timezone.utc) + info = { + "sandbox_id": sandbox_id, + "status": "pending", + "image": request.image, + "created_at": created_at.isoformat(), + "ttl_seconds": effective_ttl, + "expires_at": (created_at + timedelta(seconds=effective_ttl)).isoformat(), + "network": request.network, + "labels": request.labels, + "error": None, + } + return JSONResponse(status_code=202, content=info) + + @v1.get("/sandboxes", response_model=SandboxList) + async def list_sandboxes( + label: List[str] = Query(default=[]), + ) -> Dict[str, Any]: + filters = _parse_label_filters(label) + named = [(name, resolver.get(name)) for name in resolver.list_names()] + # Concurrently: sequential describes would make the listing scale + # linearly with the number of sandboxes. + results = await asyncio.gather( + *( + _describe_or_pending(name, handle) + for name, handle in named + if handle is not None + ), + return_exceptions=True, + ) + sandboxes: List[Dict[str, Any]] = [] + for info in results: + if isinstance(info, _ApiError) and info.code == "sandbox_not_found": + continue + if isinstance(info, BaseException): + raise info + if all(info.get("labels", {}).get(k) == v for k, v in filters.items()): + sandboxes.append(info) + return {"sandboxes": sandboxes} + + @v1.get("/sandboxes/{sandbox_id}", response_model=SandboxInfo) + async def get_sandbox( + sandbox_id: str, wait_seconds: float = _WAIT_QUERY + ) -> Dict[str, Any]: + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + return await _describe_or_pending(sandbox_id, handle, wait_seconds) + + @v1.delete("/sandboxes/{sandbox_id}") + async def delete_sandbox(sandbox_id: str) -> Dict[str, str]: + handle = resolver.get(sandbox_id) + if handle is not None: + try: + await handle.terminate.remote() + except Exception as exc: + if not _is_actor_gone(exc): + raise + resolver.kill(handle) + # Idempotent: deleting an unknown or already-gone sandbox succeeds. + return {"sandbox_id": sandbox_id, "status": "terminated"} + + # ------------------------------------------------------------------ + # Execs + # ------------------------------------------------------------------ + + @v1.post( + "/sandboxes/{sandbox_id}/execs", + response_model=ExecStarted, + status_code=202, + ) + async def start_exec(sandbox_id: str, request: StartExecRequest) -> Dict[str, Any]: + if ( + request.timeout_seconds is not None + and request.timeout_seconds > settings.max_exec_timeout_seconds + ): + raise _ApiError( + 400, + "invalid_request", + f"timeout_seconds may not exceed {settings.max_exec_timeout_seconds}", + ) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, + handle.start_exec.remote( + command=request.command, + cwd=request.cwd, + env=request.env, + timeout_seconds=request.timeout_seconds, + shell=request.shell, + user=request.user, + ), + ) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + return result + + @v1.get("/sandboxes/{sandbox_id}/execs/{exec_id}", response_model=ExecInfo) + async def get_exec( + sandbox_id: str, exec_id: str, wait_seconds: float = _WAIT_QUERY + ) -> Dict[str, Any]: + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, + handle.get_exec.remote(exec_id, wait_seconds=wait_seconds), + extra_wait=wait_seconds, + ) + if result.get("error_code") == "exec_not_found": + raise _ApiError(404, "exec_not_found", result["message"]) + return result + + # ------------------------------------------------------------------ + # Files + # ------------------------------------------------------------------ + + def _validate_file_path(path: str) -> None: + if not path.startswith("/"): + raise _ApiError(400, "invalid_request", "file path must be absolute") + + @v1.put("/sandboxes/{sandbox_id}/files", status_code=204) + async def put_file( + sandbox_id: str, + request: Request, + path: str = Query(min_length=1), + append: bool = Query( + default=False, + description=( + "Append to the file instead of truncating it. Lets clients " + "chunk large uploads under proxy body-size limits." + ), + ), + ) -> Response: + _validate_file_path(path) + body = await request.body() + if len(body) > settings.max_file_bytes: + raise _ApiError( + 413, + "payload_too_large", + f"file body may not exceed {settings.max_file_bytes} bytes", + ) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, handle.write_file.remote(path, body, append=append) + ) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + return Response(status_code=204) + + @v1.get("/sandboxes/{sandbox_id}/files") + async def get_file(sandbox_id: str, path: str = Query(min_length=1)) -> Response: + _validate_file_path(path) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call(sandbox_id, handle.read_file.remote(path)) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + if result.get("error_code") == "file_not_found": + raise _ApiError(404, "file_not_found", result["message"]) + return Response( + content=result["content"], media_type="application/octet-stream" + ) + + app = FastAPI(title="Ray Sandbox API", version="1.0.0") + app.include_router(public) + app.include_router(v1) + + @app.exception_handler(_ApiError) + async def api_error_handler(request: Request, exc: _ApiError) -> JSONResponse: + return JSONResponse( + status_code=exc.status_code, + content={"error": {"code": exc.code, "message": exc.message}}, + ) + + @app.exception_handler(Exception) + async def internal_error_handler(request: Request, exc: Exception) -> JSONResponse: + logger.exception("Unhandled error serving %s", request.url.path) + return JSONResponse( + status_code=500, + content={"error": {"code": "internal", "message": "internal server error"}}, + ) + + return app + + +@PublicAPI(stability="alpha") +def build_app(args: Optional[Dict[str, Any]] = None) -> Any: + """Ray Serve application builder. + + Usable directly with ``serve run`` (builder args become + :class:`SandboxAPISettings` fields):: + + serve run ray.experimental.sandbox.http.app:build_app + + or from an Anyscale service config via ``import_path``. + """ + from ray import serve + + settings = SandboxAPISettings(**(args or {})) + fastapi_app = create_app(settings) + + # The API is long-poll based (requests deliberately hold a slot for up + # to ~30s), so the replica must admit far more concurrent requests than + # Serve's default allows. + @serve.deployment(name="RaySandboxAPI", max_ongoing_requests=1000) + @serve.ingress(fastapi_app) + class SandboxAPIIngress: + pass + + return SandboxAPIIngress.options(num_replicas=settings.num_replicas).bind() diff --git a/python/ray/experimental/sandbox/http/host.py b/python/ray/experimental/sandbox/http/host.py new file mode 100644 index 000000000000..95d5c536ea6c --- /dev/null +++ b/python/ray/experimental/sandbox/http/host.py @@ -0,0 +1,522 @@ +"""Per-sandbox actor backing the Ray Sandbox HTTP API. + +Each API sandbox is one ``SandboxHost``, created by the HTTP layer as a +*named, detached* Ray actor (name = sandbox id). The detached actors are the +API's registry: any Serve replica resolves a sandbox with ``ray.get_actor`` +and the service keeps no state of its own, so replicas can restart or scale +without losing sandboxes. + +``SandboxHost`` composes :class:`~ray.experimental.sandbox.runtime.SandboxRuntime` +rather than the ``ray.experimental.sandbox.Sandbox`` actor because the API +needs behavior the upstream actor does not provide: + +* **Boot in the background** — the upstream actor pulls the image and boots + the container inside ``__init__``, so creation errors only surface on the + first method call. Here ``__init__`` is trivial and ``boot()`` runs as a + background task, making progress (``pending -> pulling -> starting -> + running``) and failures pollable over HTTP. +* **Exec as jobs** — commands can outrun any HTTP request (and the load + balancers in front of an Anyscale service), so ``start_exec`` returns an id + immediately and results are polled. +* **A TTL that reclaims everything** — the upstream TTL timer deletes the + sandbox but leaks the hosting actor and its resource reservation; this one + deletes the sandbox and then kills its own actor. + +Capability grants and host-network behavior (netns, resolv.conf) are plain +``SandboxConfig`` fields handled by the core runtime; nothing is patched here. + +Cross-actor control flow uses plain dicts (``{"error_code": ...}``) instead +of exceptions: Ray re-raises remote exceptions as dynamically-built +``RayTaskError`` subclasses, which makes matching them in the HTTP layer +fragile. Unexpected exceptions still propagate and map to HTTP 500. +""" + +import asyncio +import logging +import os +import platform +import shutil +import urllib.request +import uuid +from collections import OrderedDict +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Dict, List, Optional, Tuple, Union + +from ray.experimental.sandbox.exceptions import SandboxError, SandboxTimeoutError +from ray.experimental.sandbox.http.schemas import DOCKER_DEFAULT_CAPABILITIES +from ray.util.annotations import DeveloperAPI + +logger = logging.getLogger(__name__) + +_TERMINAL_EXEC_STATUSES = ("completed", "timeout", "error") +_BOOTING_STATUSES = ("pending", "pulling", "starting") + + +def _truncate_output(text: str, max_bytes: int) -> Tuple[str, bool]: + """Cap *text* at *max_bytes* of UTF-8, with a loud trailing marker.""" + data = text.encode("utf-8", errors="replace") + if len(data) <= max_bytes: + return text, False + clipped = data[:max_bytes].decode("utf-8", errors="replace") + return ( + clipped + f"\n[truncated by ray-sandbox: output exceeded {max_bytes} bytes]", + True, + ) + + +def _ensure_runsc_installed() -> None: + """Download runsc from the official gVisor release bucket onto this node. + + Opt-in via the ``auto_install_runsc`` server setting, for running the + service on stock node images. The download lands in a temp directory + prepended to this process's PATH, which the runsc subprocesses inherit. + """ + if shutil.which("runsc"): + return + # One shared, cached location per node: repeated boots reuse it instead + # of leaking a ~40MB download per sandbox, and concurrent downloaders + # converge through the atomic rename. + shared_bin = "/tmp/ray-sandbox-runsc" + runsc_path = os.path.join(shared_bin, "runsc") + if not os.path.exists(runsc_path): + os.makedirs(shared_bin, mode=0o755, exist_ok=True) + arch = ( + "aarch64" + if platform.machine().lower() in ("aarch64", "arm64") + else "x86_64" + ) + url = ( + "https://storage.googleapis.com/gvisor/releases/release/latest/" + f"{arch}/runsc" + ) + logger.info("runsc not found on this node; downloading from %s", url) + tmp_path = f"{runsc_path}.tmp.{os.getpid()}" + urllib.request.urlretrieve(url, tmp_path) + os.chmod(tmp_path, 0o755) + os.replace(tmp_path, runsc_path) + if shared_bin not in os.environ.get("PATH", "").split(os.pathsep): + os.environ["PATH"] = f"{shared_bin}{os.pathsep}{os.environ.get('PATH', '')}" + + +class _ExecJob: + """One submitted command and its (eventual) result.""" + + def __init__(self, exec_id: str) -> None: + self.exec_id = exec_id + self.status = "running" + self.exit_code: Optional[int] = None + self.stdout: Optional[str] = None + self.stderr: Optional[str] = None + self.stdout_truncated = False + self.stderr_truncated = False + self.duration_seconds: Optional[float] = None + self.error: Optional[str] = None + self.done = asyncio.Event() + + def to_dict(self) -> Dict[str, Any]: + return { + "exec_id": self.exec_id, + "status": self.status, + "exit_code": self.exit_code, + "stdout": self.stdout, + "stderr": self.stderr, + "stdout_truncated": self.stdout_truncated, + "stderr_truncated": self.stderr_truncated, + "duration_seconds": self.duration_seconds, + "error": self.error, + } + + +@DeveloperAPI +class SandboxHost: + """Hosts one gVisor sandbox for the HTTP API. + + Instantiated by the HTTP layer via ``ray.remote(SandboxHost)`` as a named + detached async actor; unit tests instantiate it directly with a fake + runtime factory, so nothing here may assume a Ray context except the + self-destruct path (which degrades to a no-op outside an actor). + + Args: + sandbox_id: The API-level sandbox id; also the detached actor's name. + spec: Sandbox creation spec (validated request data, plain dict). + settings: Server limits (``max_output_bytes``, ``max_exec_history``). + runtime_factory: Test seam; defaults to ``SandboxRuntime``. + """ + + def __init__( + self, + sandbox_id: str, + spec: Dict[str, Any], + settings: Dict[str, Any], + runtime_factory: Optional[Callable[[], Any]] = None, + ) -> None: + self._sandbox_id = sandbox_id + self._spec = spec + self._max_output_bytes = int(settings.get("max_output_bytes", 10 * 1024**2)) + self._max_exec_history = int(settings.get("max_exec_history", 256)) + self._auto_install_runsc = bool(settings.get("auto_install_runsc", False)) + self._runtime_factory = runtime_factory + self._runtime: Optional[Any] = None + self._instance_id: Optional[str] = None + self._status = "pending" + self._error: Optional[str] = None + self._created_at = datetime.now(timezone.utc) + self._status_changed = asyncio.Event() + self._execs: "OrderedDict[str, _ExecJob]" = OrderedDict() + self._exec_tasks: Dict[str, asyncio.Task] = {} + self._ttl_task: Optional[asyncio.Task] = None + self._boot_started = False + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def boot(self) -> None: + """Pull the image and start the sandbox, recording progress. + + Fired by the HTTP layer right after actor creation and never awaited + for its result; every outcome (including failure) lands in the status + this actor reports. Idempotent so a lost-then-retried create (via + ``get_if_exists``) cannot boot twice. + """ + if self._boot_started: + return + self._boot_started = True + + ttl_seconds = self._spec.get("ttl_seconds") + if ttl_seconds is not None: + # Started before the boot work so even a sandbox stuck in a + # failing boot is eventually reclaimed. + self._ttl_task = asyncio.create_task(self._ttl_watchdog(ttl_seconds)) + + try: + if self._auto_install_runsc: + await asyncio.to_thread(_ensure_runsc_installed) + factory = self._runtime_factory + if factory is None: + from ray.experimental.sandbox.runtime import SandboxRuntime + + factory = SandboxRuntime + self._runtime = factory() + + self._set_status("pulling") + await asyncio.to_thread( + self._runtime.pull_image, + self._spec["image"], + timeout_seconds=float( + self._spec.get("image_pull_timeout_seconds", 600.0) + ), + ) + + self._set_status("starting") + capabilities = self._spec.get("capabilities") + if capabilities is None: + capabilities = list(DOCKER_DEFAULT_CAPABILITIES) + + memory_limit_mb = self._spec.get("memory_limit_mb") + cpu_limit = self._spec.get("cpu_limit") + create_kwargs: Dict[str, Any] = {} + if self._spec.get("shell") is not None: + # Omitted rather than passed as None: SandboxConfig.shell is + # a plain str with a /bin/bash default. + create_kwargs["shell"] = self._spec["shell"] + # Requests and limits are deliberately decoupled: cpu_request / + # memory_request_mb size the hosting actor (cluster scheduling) + # and only cpu_limit / memory_limit_mb become cgroup caps — + # unlike the upstream Sandbox actor, which infers a cpu quota + # from its assigned resources. + self._instance_id = await asyncio.to_thread( + self._runtime.create, + self._spec["image"], + # 0 means "no cgroup limit" to Ray Sandbox. Note that it still + # derives a cpu quota from the hosting actor's assigned CPUs + # when no explicit limit is set. + cpu=float(cpu_limit) if cpu_limit is not None else 0.0, + memory=f"{memory_limit_mb}Mi" if memory_limit_mb is not None else 0, + env=dict(self._spec.get("env") or {}), + workdir=self._spec.get("workdir"), + # Writability is the runtime's explicit contract: a scratch + # dir exists only for an explicitly passed workdir on a + # readonly rootfs; readonly=False sandboxes are fully + # writable with image WORKDIR content visible. + # The API owns the TTL (see _ttl_watchdog); the upstream + # runtime stores but never enforces this, and the upstream + # actor's timer would reclaim only the sandbox, not the actor. + ttl_seconds=None, + timeout_seconds=float(self._spec.get("start_timeout_seconds", 60.0)), + rootless=bool(self._spec.get("rootless", True)), + network=self._spec.get("network", "none"), + dns=self._spec.get("dns"), + capabilities=capabilities, + readonly=bool(self._spec.get("readonly", True)), + **create_kwargs, + ) + + self._set_status("running") + logger.info( + "Sandbox %s running (instance %s, image %s)", + self._sandbox_id, + self._instance_id, + self._spec["image"], + ) + except Exception as exc: + # The actor stays alive holding the error so clients can read it; + # DELETE or the TTL reclaims it. + logger.warning("Sandbox %s failed to boot: %s", self._sandbox_id, exc) + self._error = str(exc) + self._set_status("error") + await self._delete_sandbox_instance() + + async def _ttl_watchdog(self, ttl_seconds: float) -> None: + await asyncio.sleep(ttl_seconds) + logger.info( + "Sandbox %s reached its TTL (%ss); terminating", + self._sandbox_id, + ttl_seconds, + ) + await self._shutdown() + self._self_destruct() + + def _self_destruct(self) -> None: + """Kill this actor so the TTL reclaims its name and reservation. + + ``ray.kill`` on the self-handle (rather than ``exit_actor``) because + the watchdog runs as a self-spawned asyncio task, outside any Ray + method invocation, where ``exit_actor``'s control-flow exception has + nothing to catch it. No-op outside an actor (unit tests). + """ + try: + import ray + + handle = ray.get_runtime_context().current_actor + ray.kill(handle) + except Exception: + logger.debug( + "Sandbox %s host is not a Ray actor; skipping self-destruct", + self._sandbox_id, + ) + + async def _shutdown(self) -> None: + """Cancel work and delete the sandbox instance. Idempotent.""" + if self._ttl_task is not None and self._ttl_task is not asyncio.current_task(): + self._ttl_task.cancel() + self._ttl_task = None + for task in self._exec_tasks.values(): + task.cancel() + self._exec_tasks.clear() + for job in self._execs.values(): + if job.status == "running": + job.status = "error" + job.error = "sandbox terminated while the command was running" + job.done.set() + await self._delete_sandbox_instance() + self._set_status("terminated") + + async def _delete_sandbox_instance(self) -> None: + if self._runtime is None or self._instance_id is None: + return + instance_id, self._instance_id = self._instance_id, None + try: + await asyncio.to_thread(self._runtime.delete, instance_id) + except Exception as exc: + logger.warning("Failed to delete sandbox instance %s: %s", instance_id, exc) + + async def terminate(self) -> Dict[str, Any]: + """Delete the sandbox and mark this host terminated. + + The HTTP layer kills the actor afterwards; splitting the two keeps + this method's reply deliverable. + """ + await self._shutdown() + return {"ok": True} + + # ------------------------------------------------------------------ + # Introspection + # ------------------------------------------------------------------ + + def _set_status(self, status: str) -> None: + self._status = status + # Replace-then-set so current waiters wake once and later waiters + # block on a fresh event. + event, self._status_changed = self._status_changed, asyncio.Event() + event.set() + + def _info(self) -> Dict[str, Any]: + ttl_seconds = self._spec.get("ttl_seconds") + expires_at = ( + self._created_at + timedelta(seconds=ttl_seconds) + if ttl_seconds is not None + else None + ) + return { + "sandbox_id": self._sandbox_id, + "status": self._status, + "image": self._spec["image"], + "created_at": self._created_at.isoformat(), + "ttl_seconds": ttl_seconds, + "expires_at": expires_at.isoformat() if expires_at else None, + "network": self._spec.get("network", "none"), + "labels": dict(self._spec.get("labels") or {}), + "error": self._error, + } + + async def describe(self, wait_seconds: float = 0.0) -> Dict[str, Any]: + """Report sandbox state, optionally long-polling while it boots.""" + if wait_seconds > 0 and self._status in _BOOTING_STATUSES: + event = self._status_changed + try: + await asyncio.wait_for(event.wait(), timeout=wait_seconds) + except asyncio.TimeoutError: + pass + return self._info() + + # ------------------------------------------------------------------ + # Exec jobs + # ------------------------------------------------------------------ + + async def start_exec( + self, + command: Union[str, List[str]], + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + timeout_seconds: Optional[float] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + exec_id = f"ex-{uuid.uuid4().hex[:12]}" + job = _ExecJob(exec_id) + self._execs[exec_id] = job + self._prune_exec_history() + task = asyncio.create_task( + self._run_exec(job, command, cwd, env, timeout_seconds, shell, user) + ) + self._exec_tasks[exec_id] = task + task.add_done_callback(lambda _: self._exec_tasks.pop(exec_id, None)) + return {"exec_id": exec_id, "status": job.status} + + def _prune_exec_history(self) -> None: + # Evict oldest *finished* jobs beyond the cap; running jobs must stay + # addressable, so with a pathological number in flight the dict may + # exceed the cap rather than lose one. + finished = [ + exec_id + for exec_id, job in self._execs.items() + if job.status in _TERMINAL_EXEC_STATUSES + ] + excess = len(self._execs) - self._max_exec_history + for exec_id in finished[: max(0, excess)]: + del self._execs[exec_id] + + async def _run_exec( + self, + job: _ExecJob, + command: Union[str, List[str]], + cwd: Optional[str], + env: Optional[Dict[str, str]], + timeout_seconds: Optional[float], + shell: Optional[str], + user: Optional[str], + ) -> None: + try: + result = await self._runtime.exec_async( + self._instance_id, + command, + timeout=timeout_seconds, + cwd=cwd, + env=env or None, + shell=shell, + user=user, + ) + except asyncio.CancelledError: + # _shutdown already marked the job; just stop. + raise + except SandboxTimeoutError: + job.status = "timeout" + job.error = f"command timed out after {timeout_seconds} seconds" + except SandboxError as exc: + job.status = "error" + job.error = str(exc) + except Exception as exc: + logger.warning( + "Exec %s in sandbox %s failed unexpectedly: %s", + job.exec_id, + self._sandbox_id, + exc, + ) + job.status = "error" + job.error = str(exc) + else: + job.status = "completed" + job.exit_code = result.exit_code + job.stdout, job.stdout_truncated = _truncate_output( + result.stdout, self._max_output_bytes + ) + job.stderr, job.stderr_truncated = _truncate_output( + result.stderr, self._max_output_bytes + ) + job.duration_seconds = result.duration_seconds + finally: + if not job.done.is_set(): + job.done.set() + + async def get_exec(self, exec_id: str, wait_seconds: float = 0.0) -> Dict[str, Any]: + job = self._execs.get(exec_id) + if job is None: + return { + "error_code": "exec_not_found", + "message": f"unknown exec id {exec_id!r}", + } + if wait_seconds > 0 and job.status == "running": + try: + await asyncio.wait_for(job.done.wait(), timeout=wait_seconds) + except asyncio.TimeoutError: + pass + return job.to_dict() + + # ------------------------------------------------------------------ + # Files + # ------------------------------------------------------------------ + + async def write_file( + self, path: str, content: bytes, append: bool = False + ) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + await asyncio.to_thread( + self._runtime.write_file, + self._instance_id, + path, + content, + append, + ) + return {"ok": True} + + async def read_file(self, path: str) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + try: + content = await asyncio.to_thread( + self._runtime.read_file, self._instance_id, path + ) + except SandboxError as exc: + # Upstream read_file shells out to `cat`; a missing file is the + # overwhelmingly common failure, so report it as such. + return {"error_code": "file_not_found", "message": str(exc)} + return {"ok": True, "content": content} diff --git a/python/ray/experimental/sandbox/http/schemas.py b/python/ray/experimental/sandbox/http/schemas.py new file mode 100644 index 000000000000..ede52cee4952 --- /dev/null +++ b/python/ray/experimental/sandbox/http/schemas.py @@ -0,0 +1,306 @@ +"""Pydantic request/response models for the Ray Sandbox HTTP API (v1). + +These models are the API contract: the OpenAPI schema FastAPI generates from +them is what clients (e.g. the Harbor ``ray-sandbox`` environment) are written +against. Changing a field here is a contract change; update the snapshot test +in ``tests/test_http_schemas.py`` deliberately when you do. +""" + +from datetime import datetime +from typing import Dict, List, Literal, Optional, Union + +from pydantic import BaseModel, Field, field_validator + +# Re-exported so the wire contract documents its default in one place; the +# canonical definition (and rationale) lives in the core sandbox config. +# The API defaults to Docker's set so images behave the way they do under +# Docker; the sets are written exactly, so ``capabilities: []`` runs the +# sandbox with no capabilities at all. +from ray.experimental.sandbox.config import ( # noqa: E402 + DOCKER_DEFAULT_CAPABILITIES, + VALID_NETWORK_MODES, +) +from ray.util.annotations import PublicAPI + +SandboxStatusName = Literal[ + "pending", "pulling", "starting", "running", "error", "terminated" +] +ExecStatusName = Literal["running", "completed", "timeout", "error"] + +_MAX_LABELS = 16 +_MAX_LABEL_LENGTH = 256 + + +@PublicAPI(stability="alpha") +class ResourceSpec(BaseModel): + """Resource requests (cluster reservation) and limits (in-sandbox cgroup). + + Requests reserve capacity on the Ray cluster for the actor hosting the + sandbox; limits cap the sandbox itself via its cgroup. When only a limit + is given the request defaults to it, so a capped sandbox is also scheduled + onto capacity that can honor the cap. + + Caveat for CPU: Ray Sandbox derives a cgroup cpu quota from the hosting + actor's assigned CPUs whenever no explicit cpu limit is set, so a + request-only CPU spec also ends up capped at the requested value. Memory + has no such coupling. + """ + + cpu_request: Optional[float] = Field(default=None, gt=0) + cpu_limit: Optional[float] = Field(default=None, gt=0) + memory_request_mb: Optional[int] = Field(default=None, gt=0) + memory_limit_mb: Optional[int] = Field(default=None, gt=0) + custom: Optional[Dict[str, float]] = Field( + default=None, + description=( + "Custom Ray resources required to schedule the sandbox's hosting " + 'actor, e.g. {"gvisor": 1} to pin sandboxes to runsc-equipped nodes.' + ), + ) + + +@PublicAPI(stability="alpha") +class CreateSandboxRequest(BaseModel): + """Request body for ``POST /api/v1/sandboxes``.""" + + image: str = Field( + min_length=1, + description=( + "Container image reference (e.g. 'python:3.12-slim'). Pulled " + "anonymously from the registry; the image must be public." + ), + ) + env: Dict[str, str] = Field(default_factory=dict) + workdir: Optional[str] = Field( + default=None, + description=( + "Sandbox-level working directory; also the default cwd for execs. " + "None uses the image's own WORKDIR, like Docker." + ), + ) + ttl_seconds: Optional[int] = Field( + default=3600, + ge=1, + description=( + "Auto-cleanup TTL. The sandbox and its hosting actor are " + "terminated this many seconds after creation. null disables the " + "TTL (subject to the server's max_ttl_seconds cap)." + ), + ) + network: str = Field( + default="none", + description=( + "runsc network mode; one of " f"{', '.join(VALID_NETWORK_MODES)}." + ), + ) + dns: Optional[List[str]] = Field( + default=None, + description=( + "Nameserver IPs for a generated /etc/resolv.conf (mirroring " + "docker --dns). Defaults to public resolvers for network=" + "'public'; overrides the host file for network='host'." + ), + ) + shell: Optional[str] = Field( + default=None, + description=( + "Shell that runs string exec commands (e.g. '/bin/sh' for " + "images without bash). null uses the sandbox default, /bin/bash." + ), + ) + rootless: bool = True + readonly: bool = True + resources: Optional[ResourceSpec] = None + labels: Dict[str, str] = Field(default_factory=dict) + capabilities: Optional[List[str]] = Field( + default=None, + description=( + "Linux capabilities granted to the sandbox (the sets are " + "written exactly). null applies the server default (Docker's " + "14-capability set); [] runs with no capabilities." + ), + ) + image_pull_timeout_seconds: float = Field(default=600.0, gt=0) + start_timeout_seconds: float = Field(default=60.0, gt=0) + client_token: Optional[str] = Field( + default=None, + min_length=1, + max_length=128, + description=( + "Idempotency token. Creates with the same token resolve to the " + "same sandbox, so a client that lost a create response can retry " + "without leaking a duplicate sandbox." + ), + ) + + @field_validator("network") + @classmethod + def _validate_network(cls, network: str) -> str: + # Sourced from the core sandbox config so a new Ray mode is accepted + # here without an API change. + if network not in VALID_NETWORK_MODES: + raise ValueError(f"network must be one of {VALID_NETWORK_MODES}") + return network + + @field_validator("labels") + @classmethod + def _validate_labels(cls, labels: Dict[str, str]) -> Dict[str, str]: + if len(labels) > _MAX_LABELS: + raise ValueError(f"at most {_MAX_LABELS} labels are allowed") + for key, value in labels.items(): + if not key or len(key) > _MAX_LABEL_LENGTH: + raise ValueError(f"label keys must be 1-{_MAX_LABEL_LENGTH} characters") + if len(value) > _MAX_LABEL_LENGTH: + raise ValueError( + f"label values must be at most {_MAX_LABEL_LENGTH} characters" + ) + return labels + + +@PublicAPI(stability="alpha") +class SandboxInfo(BaseModel): + """Sandbox state as reported by ``GET /api/v1/sandboxes/{id}``.""" + + sandbox_id: str + status: SandboxStatusName + image: str + created_at: datetime + ttl_seconds: Optional[int] = None + expires_at: Optional[datetime] = None + network: str + labels: Dict[str, str] = Field(default_factory=dict) + error: Optional[str] = Field( + default=None, description="Failure detail when status is 'error'." + ) + + +@PublicAPI(stability="alpha") +class SandboxList(BaseModel): + """Response body for ``GET /api/v1/sandboxes``.""" + + sandboxes: List[SandboxInfo] + + +@PublicAPI(stability="alpha") +class StartExecRequest(BaseModel): + """Request body for ``POST /api/v1/sandboxes/{id}/execs``. + + A string command runs under the sandbox's shell (``/bin/bash`` unless + the sandbox or this request configures another); a list is executed + argv-style without a shell. + """ + + command: Union[str, List[str]] + cwd: Optional[str] = None + env: Dict[str, str] = Field(default_factory=dict) + timeout_seconds: Optional[float] = Field(default=None, gt=0) + shell: Optional[str] = Field( + default=None, + description="Override the sandbox's shell for this command only.", + ) + user: Optional[str] = Field( + default=None, + description=( + "Run as this user: a numeric uid, 'uid:gid', or a name from " + "the image's /etc/passwd. Default: the image's user." + ), + ) + + @field_validator("command") + @classmethod + def _validate_command(cls, command: Union[str, List[str]]) -> Union[str, List[str]]: + if isinstance(command, str): + if not command.strip(): + raise ValueError("command must be non-empty") + elif not command or not all(isinstance(part, str) for part in command): + raise ValueError("argv command must be a non-empty list of strings") + return command + + +@PublicAPI(stability="alpha") +class ExecStarted(BaseModel): + """Response body for ``POST /api/v1/sandboxes/{id}/execs``.""" + + exec_id: str + status: ExecStatusName + + +@PublicAPI(stability="alpha") +class ExecInfo(BaseModel): + """Exec state as reported by ``GET /api/v1/sandboxes/{id}/execs/{exec_id}``.""" + + exec_id: str + status: ExecStatusName + exit_code: Optional[int] = None + stdout: Optional[str] = None + stderr: Optional[str] = None + stdout_truncated: bool = False + stderr_truncated: bool = False + duration_seconds: Optional[float] = None + error: Optional[str] = Field( + default=None, description="Failure detail when status is 'timeout' or 'error'." + ) + + +@PublicAPI(stability="alpha") +class SandboxAPISettings(BaseModel): + """Server-side settings, passed as Serve application builder args.""" + + namespace: str = Field( + default="ray_sandbox_api", + description="Ray namespace holding the detached per-sandbox actors.", + ) + token_env_var: str = Field( + default="RAY_SANDBOX_API_TOKEN", + description=( + "Environment variable read at app construction for the bearer " + "token. Unset/empty disables the app-level check (an Anyscale " + "service already enforces its own bearer token at the platform " + "edge)." + ), + ) + num_replicas: int = Field(default=1, ge=1) + max_output_bytes: int = Field( + default=10 * 1024 * 1024, + gt=0, + description="Per-stream cap on retained exec stdout/stderr.", + ) + max_file_bytes: int = Field( + default=256 * 1024 * 1024, + gt=0, + description="Cap on file upload body size (HTTP 413 above it).", + ) + max_ttl_seconds: int = Field(default=7 * 24 * 3600, gt=0) + max_exec_timeout_seconds: float = Field(default=6 * 3600.0, gt=0) + max_exec_history: int = Field( + default=256, + gt=0, + description="Completed exec records retained per sandbox.", + ) + scheduling_grace_seconds: float = Field( + default=15.0, + gt=0, + description=( + "How long a request may wait on a sandbox whose hosting actor " + "has not been scheduled yet (e.g. the cluster is scaling up) " + "before reporting the sandbox as pending instead of blocking." + ), + ) + default_actor_num_cpus: float = Field( + default=1.0, + ge=0, + description="Actor CPU reservation when the request has no cpu_request.", + ) + default_capabilities: List[str] = Field( + default_factory=lambda: list(DOCKER_DEFAULT_CAPABILITIES) + ) + auto_install_runsc: bool = Field( + default=False, + description=( + "Download gVisor's runsc from the official release bucket onto " + "any node that lacks it, at first sandbox boot. Lets the service " + "run on stock images (e.g. an unmodified Anyscale cluster image) " + "at the cost of a one-time ~40MB download per node. Prefer " + "baking runsc into the node image for production." + ), + ) diff --git a/python/ray/experimental/sandbox/http/tests/__init__.py b/python/ray/experimental/sandbox/http/tests/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/python/ray/experimental/sandbox/http/tests/conftest.py b/python/ray/experimental/sandbox/http/tests/conftest.py new file mode 100644 index 000000000000..dbe72807693f --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/conftest.py @@ -0,0 +1,267 @@ +"""Fixtures for the Ray Sandbox HTTP API tests. + +Unlike ``sandbox/tests/conftest.py`` there is no TEST_SANDBOX gate here: the +unit tests fake the sandbox runtime and the actor layer, so they run on any +platform with no cluster and no runsc. Only ``test_http_integration.py`` +needs the real thing and gates itself. +""" + +import asyncio +import os +import platform +import shutil +import tempfile +import threading +import time +import urllib.request +from typing import Any, Dict, List, Optional, Union + +import pytest + +from ray.experimental.sandbox.exceptions import ( + SandboxExecError, + SandboxTimeoutError, +) +from ray.experimental.sandbox.http.host import SandboxHost + + +def _sandbox_test_enabled() -> bool: + try: + from ray._private.test_utils import sandbox_test_enabled + except ImportError: + return os.environ.get("TEST_SANDBOX") == "1" + return sandbox_test_enabled() + + +class FakeExecResult: + def __init__( + self, + exit_code: int = 0, + stdout: str = "", + stderr: str = "", + duration_seconds: float = 0.01, + ) -> None: + self.exit_code = exit_code + self.stdout = stdout + self.stderr = stderr + self.duration_seconds = duration_seconds + + +class FakeSandboxRuntime: + """Scriptable stand-in for ``ray.experimental.sandbox.SandboxRuntime``.""" + + def __init__(self) -> None: + self.instance_id = "ray-sandbox-fake0001" + self.pull_calls: List[Dict[str, Any]] = [] + self.create_calls: List[Dict[str, Any]] = [] + self.exec_calls: List[Dict[str, Any]] = [] + self.written_files: Dict[str, bytes] = {} + self.readable_files: Dict[str, bytes] = {} + self.deleted: List[str] = [] + self.pull_error: Optional[Exception] = None + self.create_error: Optional[Exception] = None + self.exec_error: Optional[Exception] = None + self.exec_results: List[FakeExecResult] = [] + # When set, pull_image blocks until the event fires, holding the + # sandbox in the "pulling" state for conflict tests. + self.pull_gate: Optional[threading.Event] = None + + def pull_image(self, image: str, timeout_seconds: float = 120.0) -> str: + self.pull_calls.append({"image": image, "timeout_seconds": timeout_seconds}) + if self.pull_gate is not None: + if not self.pull_gate.wait(timeout=30): + raise RuntimeError("pull gate never released") + if self.pull_error is not None: + raise self.pull_error + return f"/tmp/fake-images/{image}" + + def create(self, image: str, **kwargs: Any) -> str: + self.create_calls.append({"image": image, **kwargs}) + if self.create_error is not None: + raise self.create_error + return self.instance_id + + def exec( + self, + instance_id: str, + command: Union[str, List[str]], + timeout: Optional[float] = None, + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> FakeExecResult: + self.exec_calls.append( + { + "instance_id": instance_id, + "command": command, + "timeout": timeout, + "cwd": cwd, + "env": env, + "shell": shell, + "user": user, + } + ) + if self.exec_error is not None: + raise self.exec_error + if self.exec_results: + return self.exec_results.pop(0) + return FakeExecResult() + + async def exec_async( + self, + instance_id: str, + command: Union[str, List[str]], + timeout: Optional[float] = None, + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> FakeExecResult: + return await asyncio.to_thread( + self.exec, + instance_id, + command, + timeout=timeout, + cwd=cwd, + env=env, + shell=shell, + user=user, + ) + + def write_file( + self, + instance_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, + ) -> None: + if isinstance(content, str): + content = content.encode("utf-8") + if append and path in self.written_files: + self.written_files[path] += content + else: + self.written_files[path] = content + + def read_file(self, instance_id: str, path: str) -> bytes: + if path not in self.readable_files: + raise SandboxExecError(f"cat: {path}: No such file or directory") + return self.readable_files[path] + + def delete(self, instance_id: str) -> None: + self.deleted.append(instance_id) + + +class _FakeRemoteMethod: + """Mimics ``handle.method.remote(...)``: schedules eagerly, returns an awaitable.""" + + def __init__(self, fn: Any) -> None: + self._fn = fn + + def remote(self, *args: Any, **kwargs: Any) -> "asyncio.Task": + return asyncio.get_running_loop().create_task(self._fn(*args, **kwargs)) + + +class FakeHandle: + def __init__(self, host: SandboxHost) -> None: + self.host = host + + def __getattr__(self, name: str) -> _FakeRemoteMethod: + return _FakeRemoteMethod(getattr(self.host, name)) + + +class FakeResolver: + """In-process stand-in for ``RayActorHandleResolver``.""" + + def __init__(self) -> None: + self.handles: Dict[str, FakeHandle] = {} + self.runtimes: List[FakeSandboxRuntime] = [] + self.create_options: List[Dict[str, Any]] = [] + self.killed: List[str] = [] + # Applied to the next runtime a created host builds. + self.next_runtime: Optional[FakeSandboxRuntime] = None + + def _runtime_factory(self) -> FakeSandboxRuntime: + runtime = self.next_runtime or FakeSandboxRuntime() + self.next_runtime = None + self.runtimes.append(runtime) + return runtime + + def create( + self, + name: str, + actor_options: Dict[str, Any], + ctor_kwargs: Dict[str, Any], + ) -> FakeHandle: + self.create_options.append({"name": name, **actor_options}) + # get_if_exists semantics: racing creates converge on one host. + if name in self.handles: + return self.handles[name] + host = SandboxHost(runtime_factory=self._runtime_factory, **ctor_kwargs) + handle = FakeHandle(host) + self.handles[name] = handle + return handle + + def get(self, name: str) -> Optional[FakeHandle]: + return self.handles.get(name) + + def list_names(self) -> List[str]: + return list(self.handles) + + def kill(self, handle: FakeHandle) -> None: + for name, existing in list(self.handles.items()): + if existing is handle: + del self.handles[name] + self.killed.append(name) + + +@pytest.fixture +def fake_resolver() -> FakeResolver: + return FakeResolver() + + +@pytest.fixture(scope="session", autouse=True) +def ensure_runsc(): + """Provision runsc for the integration test, mirroring sandbox/tests.""" + if not _sandbox_test_enabled(): + return + + os.environ["RAY_SANDBOX_IGNORE_CGROUPS"] = "1" + + if not shutil.which("runsc"): + temp_bin = tempfile.mkdtemp() + os.chmod(temp_bin, 0o755) + runsc_path = os.path.join(temp_bin, "runsc") + arch = ( + "aarch64" + if platform.machine().lower() in ("aarch64", "arm64") + else "x86_64" + ) + url = f"https://storage.googleapis.com/gvisor/releases/release/latest/{arch}/runsc" + try: + urllib.request.urlretrieve(url, runsc_path) + os.chmod(runsc_path, 0o755) + os.environ["PATH"] = f"{temp_bin}:{os.environ.get('PATH', '')}" + except Exception as e: + pytest.skip(f"Failed to install runsc for sandbox tests: {e}") + + +def wait_until(predicate, timeout: float = 10.0, interval: float = 0.02) -> None: + """Poll *predicate* until true or fail the test.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(interval) + raise AssertionError("condition not met within timeout") + + +__all__ = [ + "FakeExecResult", + "FakeSandboxRuntime", + "FakeHandle", + "FakeResolver", + "SandboxExecError", + "SandboxTimeoutError", + "wait_until", +] diff --git a/python/ray/experimental/sandbox/http/tests/test_http_app.py b/python/ray/experimental/sandbox/http/tests/test_http_app.py new file mode 100644 index 000000000000..2fd389848883 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_app.py @@ -0,0 +1,684 @@ +"""HTTP-layer tests: FastAPI app + fake resolver + real SandboxHost logic. + +No Ray cluster and no runsc: the resolver seam replaces actor creation with +in-process ``SandboxHost`` instances driven by ``FakeSandboxRuntime``. The +``TestClient`` context manager keeps one event loop alive across requests so +background boot tasks progress between calls, exactly like a live server. +""" + +import sys + +import pytest +from fastapi.testclient import TestClient + +from ray.experimental.sandbox.http.app import create_app +from ray.experimental.sandbox.http.schemas import SandboxAPISettings +from ray.experimental.sandbox.http.tests.conftest import ( + FakeExecResult, + FakeResolver, + FakeSandboxRuntime, +) + +BASE = "/api/v1" + + +def _client(resolver: FakeResolver, settings: SandboxAPISettings = None) -> TestClient: + app = create_app(settings or SandboxAPISettings(), handle_resolver=resolver) + return TestClient(app) + + +def _create_sandbox(client: TestClient, **overrides) -> dict: + body = {"image": "python:3.12-slim", "readonly": False, **overrides} + response = client.post(f"{BASE}/sandboxes", json=body) + assert response.status_code == 202, response.text + return response.json() + + +def _wait_running(client: TestClient, sandbox_id: str) -> dict: + for _ in range(100): + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}", params={"wait_seconds": 1} + ) + assert response.status_code == 200, response.text + info = response.json() + if info["status"] not in ("pending", "pulling", "starting"): + return info + raise AssertionError("sandbox never left its boot states") + + +# ---------------------------------------------------------------------- +# Auth +# ---------------------------------------------------------------------- + + +def test_bearer_auth_enforced_when_token_configured( + fake_resolver: FakeResolver, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("RAY_SANDBOX_API_TOKEN", "sekret") + with _client(fake_resolver) as client: + # Health stays public. + assert client.get(f"{BASE}/health").status_code == 200 + + response = client.get(f"{BASE}/sandboxes") + assert response.status_code == 401 + assert response.json()["error"]["code"] == "unauthorized" + + response = client.get( + f"{BASE}/sandboxes", headers={"Authorization": "Bearer wrong"} + ) + assert response.status_code == 401 + + response = client.get( + f"{BASE}/sandboxes", headers={"Authorization": "Bearer sekret"} + ) + assert response.status_code == 200 + + +def test_auth_disabled_without_token( + fake_resolver: FakeResolver, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("RAY_SANDBOX_API_TOKEN", raising=False) + with _client(fake_resolver) as client: + assert client.get(f"{BASE}/sandboxes").status_code == 200 + + +# ---------------------------------------------------------------------- +# Sandbox lifecycle +# ---------------------------------------------------------------------- + + +def test_create_then_poll_until_running(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client, labels={"team": "eval"}) + # The boot task starts immediately, so the 202 body reports whatever + # early state it reached — any pre-terminal status is legitimate. + assert info["status"] in ("pending", "pulling", "starting", "running") + assert info["sandbox_id"].startswith("sb-") + assert info["image"] == "python:3.12-slim" + assert info["labels"] == {"team": "eval"} + + running = _wait_running(client, info["sandbox_id"]) + assert running["status"] == "running" + assert running["error"] is None + + (runtime,) = fake_resolver.runtimes + assert runtime.create_calls[0]["readonly"] is False + + +def test_create_maps_resources_to_actor_options_and_limits( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox( + client, + resources={ + "cpu_request": 0.5, + "cpu_limit": 2.0, + "memory_request_mb": 256, + "memory_limit_mb": 2048, + "custom": {"gvisor": 1.0}, + }, + ) + _wait_running(client, info["sandbox_id"]) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 0.5 + assert options["memory"] == 256 * 1024 * 1024 + assert options["resources"] == {"gvisor": 1.0} + + (runtime,) = fake_resolver.runtimes + create_call = runtime.create_calls[0] + assert create_call["cpu"] == 2.0 + assert create_call["memory"] == "2048Mi" + + +def test_create_defaults_request_to_limit(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + _create_sandbox(client, resources={"cpu_limit": 4.0, "memory_limit_mb": 512}) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 4.0 + assert options["memory"] == 512 * 1024 * 1024 + + +def test_create_default_actor_cpus_without_resources( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + _create_sandbox(client) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 1.0 + assert "memory" not in options + + +def test_create_ttl_over_cap_rejected(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_ttl_seconds=100) + with _client(fake_resolver, settings) as client: + response = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "ttl_seconds": 101}, + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_create_null_ttl_clamped_to_server_cap( + fake_resolver: FakeResolver, +) -> None: + settings = SandboxAPISettings(max_ttl_seconds=1234) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client, ttl_seconds=None) + assert info["ttl_seconds"] == 1234 + + +def test_create_missing_image_is_422(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + assert client.post(f"{BASE}/sandboxes", json={}).status_code == 422 + + +def test_client_token_makes_create_idempotent( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + first = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "client_token": "trial-42"}, + ) + assert first.status_code == 202 + second = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "client_token": "trial-42"}, + ) + assert second.status_code == 200 + assert second.json()["sandbox_id"] == first.json()["sandbox_id"] + assert len(fake_resolver.runtimes) == 1 + + +def test_get_unknown_sandbox_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + response = client.get(f"{BASE}/sandboxes/sb-missing") + assert response.status_code == 404 + assert response.json()["error"]["code"] == "sandbox_not_found" + + +def test_delete_is_idempotent_and_kills_actor( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.delete(f"{BASE}/sandboxes/{sandbox_id}") + assert response.status_code == 200 + assert response.json() == { + "sandbox_id": sandbox_id, + "status": "terminated", + } + assert fake_resolver.killed == [sandbox_id] + (runtime,) = fake_resolver.runtimes + assert runtime.deleted == [runtime.instance_id] + + # Gone now, and deleting again still succeeds. + assert client.get(f"{BASE}/sandboxes/{sandbox_id}").status_code == 404 + assert client.delete(f"{BASE}/sandboxes/{sandbox_id}").status_code == 200 + + +def test_boot_failure_surfaces_as_error_status( + fake_resolver: FakeResolver, +) -> None: + runtime = FakeSandboxRuntime() + runtime.pull_error = RuntimeError("no such image") + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + final = _wait_running(client, info["sandbox_id"]) + assert final["status"] == "error" + assert "no such image" in final["error"] + + +def test_list_sandboxes_with_label_filter(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + a = _create_sandbox(client, labels={"job": "j1", "role": "env"}) + b = _create_sandbox(client, labels={"job": "j2"}) + + response = client.get(f"{BASE}/sandboxes") + assert response.status_code == 200 + assert {s["sandbox_id"] for s in response.json()["sandboxes"]} == { + a["sandbox_id"], + b["sandbox_id"], + } + + response = client.get(f"{BASE}/sandboxes", params=[("label", "job=j1")]) + assert [s["sandbox_id"] for s in response.json()["sandboxes"]] == [ + a["sandbox_id"] + ] + + response = client.get(f"{BASE}/sandboxes", params=[("label", "nonsense")]) + assert response.status_code == 400 + + +# ---------------------------------------------------------------------- +# Execs +# ---------------------------------------------------------------------- + + +def test_exec_submit_and_poll(fake_resolver: FakeResolver) -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [FakeExecResult(exit_code=3, stdout="done")] + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", + json={ + "command": ["bash", "-c", "exit 3"], + "cwd": "/app", + "env": {"A": "1"}, + "timeout_seconds": 5, + }, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + assert exec_id.startswith("ex-") + + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/execs/{exec_id}", + params={"wait_seconds": 10}, + ) + assert response.status_code == 200 + result = response.json() + assert result["status"] == "completed" + assert result["exit_code"] == 3 + assert result["stdout"] == "done" + + exec_call = runtime.exec_calls[-1] + assert exec_call["command"] == ["bash", "-c", "exit 3"] + assert exec_call["cwd"] == "/app" + assert exec_call["env"] == {"A": "1"} + assert exec_call["timeout"] == 5 + + +def test_exec_on_unknown_sandbox_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + response = client.post( + f"{BASE}/sandboxes/sb-missing/execs", json={"command": "true"} + ) + assert response.status_code == 404 + + +def test_unknown_exec_id_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.get(f"{BASE}/sandboxes/{sandbox_id}/execs/ex-nope") + assert response.status_code == 404 + assert response.json()["error"]["code"] == "exec_not_found" + + +def test_exec_while_booting_is_409(fake_resolver: FakeResolver) -> None: + import threading + + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": "true"} + ) + assert response.status_code == 409 + assert response.json()["error"]["code"] == "conflict" + + runtime.pull_gate.set() + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": "true"} + ) + assert response.status_code == 202 + + +def test_exec_timeout_over_cap_rejected(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_exec_timeout_seconds=10.0) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", + json={"command": "true", "timeout_seconds": 11}, + ) + assert response.status_code == 400 + + +def test_empty_command_is_422(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": " "} + ) + assert response.status_code == 422 + + +# ---------------------------------------------------------------------- +# Files +# ---------------------------------------------------------------------- + + +def test_file_put_and_get_roundtrip(fake_resolver: FakeResolver) -> None: + runtime = FakeSandboxRuntime() + runtime.readable_files["/data/report.bin"] = b"\x00\x01binary" + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/data/in.bin"}, + content=b"payload-bytes", + ) + assert response.status_code == 204 + assert runtime.written_files["/data/in.bin"] == b"payload-bytes" + + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/data/report.bin"}, + ) + assert response.status_code == 200 + assert response.content == b"\x00\x01binary" + assert response.headers["content-type"].startswith("application/octet-stream") + + +def test_file_get_missing_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/files", params={"path": "/nope"} + ) + assert response.status_code == 404 + assert response.json()["error"]["code"] == "file_not_found" + + +def test_file_put_too_large_413(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_file_bytes=8) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/big"}, + content=b"123456789", + ) + assert response.status_code == 413 + assert response.json()["error"]["code"] == "payload_too_large" + + +def test_file_relative_path_400(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "relative/path"}, + content=b"x", + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_file_ops_while_booting_409(fake_resolver: FakeResolver) -> None: + import threading + + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + fake_resolver.next_runtime = runtime + try: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/x"}, + content=b"x", + ) + assert response.status_code == 409 + finally: + runtime.pull_gate.set() + + +class _UnschedulableHandle: + """Mimics a handle whose actor Ray reports as permanently unschedulable.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + async def _raise(): + exc_type = type("ActorUnschedulableError", (Exception,), {}) + raise exc_type("resource shapes cannot fit the cluster") + + return asyncio.get_running_loop().create_task(_raise()) + + return _Method() + + +def test_unschedulable_actor_maps_to_409(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-unsched0001"] = _UnschedulableHandle() + + response = client.get(f"{BASE}/sandboxes/sb-unsched0001") + + assert response.status_code == 409, response.text + assert response.json()["error"]["code"] == "unschedulable" + + +class _StalledHandle: + """Mimics a handle to a created-but-unscheduled detached actor: every + remote call returns an awaitable that never resolves.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + return asyncio.get_running_loop().create_future() + + return _Method() + + +def _fast_settings() -> SandboxAPISettings: + return SandboxAPISettings(scheduling_grace_seconds=0.2) + + +def test_get_sandbox_reports_pending_while_actor_is_scheduling( + fake_resolver: FakeResolver, +) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.get(f"{BASE}/sandboxes/sb-stalled0001") + + assert response.status_code == 200, response.text + info = response.json() + assert info["status"] == "pending" + assert info["sandbox_id"] == "sb-stalled0001" + + +def test_exec_on_scheduling_actor_is_409(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.post( + f"{BASE}/sandboxes/sb-stalled0001/execs", json={"command": "echo hi"} + ) + + assert response.status_code == 409, response.text + assert "scheduled" in response.json()["error"]["message"] + + +def test_list_includes_scheduling_sandboxes_as_pending( + fake_resolver: FakeResolver, +) -> None: + client = _client(fake_resolver, _fast_settings()) + _create_sandbox(client) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.get(f"{BASE}/sandboxes") + + assert response.status_code == 200, response.text + statuses = {s["sandbox_id"]: s["status"] for s in response.json()["sandboxes"]} + assert statuses["sb-stalled0001"] == "pending" + + +def test_shell_passthrough_to_runtime(fake_resolver: FakeResolver) -> None: + """The create-level and per-exec shell fields reach the sandbox runtime.""" + with _client(fake_resolver) as client: + info = _create_sandbox(client, shell="/bin/sh") + _wait_running(client, info["sandbox_id"]) + + response = client.post( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs", + json={"command": "echo hi", "shell": "/bin/dash"}, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + result = client.get( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs/{exec_id}", + params={"wait_seconds": 5}, + ).json() + assert result["status"] == "completed" + + runtime = fake_resolver.runtimes[0] + assert runtime.create_calls[0]["shell"] == "/bin/sh" + assert runtime.exec_calls[0]["shell"] == "/bin/dash" + + +def test_exec_user_passthrough(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + _wait_running(client, info["sandbox_id"]) + + response = client.post( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs", + json={"command": "id", "user": "1000:1000"}, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + client.get( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs/{exec_id}", + params={"wait_seconds": 5}, + ) + assert fake_resolver.runtimes[0].exec_calls[0]["user"] == "1000:1000" + + +def test_invalid_network_mode_is_422(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver) + response = client.post( + f"{BASE}/sandboxes", json={"image": "python:3.12", "network": "bridge"} + ) + assert response.status_code == 422 + + +def test_chunked_file_upload_via_append(fake_resolver: FakeResolver) -> None: + """PUT ?append=true extends the file, so clients can chunk large uploads + under proxy body-size limits.""" + with _client(fake_resolver) as client: + info = _create_sandbox(client) + _wait_running(client, info["sandbox_id"]) + base = f"{BASE}/sandboxes/{info['sandbox_id']}/files" + + assert ( + client.put(base, params={"path": "/big"}, content=b"aaa").status_code == 204 + ) + assert ( + client.put( + base, params={"path": "/big", "append": "true"}, content=b"bbb" + ).status_code + == 204 + ) + + runtime = fake_resolver.runtimes[0] + assert runtime.written_files["/big"] == b"aaabbb" + + +class _UnavailableHandle: + """Mimics a handle whose actor Ray reports as *transiently* unavailable (a + restart or network blip). ``ActorUnavailableError`` subclasses + ``RayActorError``, so the app must map it to 503 rather than the 404 it + maps a genuine death to — otherwise a client_token retry during the blip + is misread as "gone" and kills a live sandbox.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + async def _raise(): + ray_actor_error = type("RayActorError", (Exception,), {}) + exc_type = type("ActorUnavailableError", (ray_actor_error,), {}) + raise exc_type("actor is restarting") + + return asyncio.get_running_loop().create_task(_raise()) + + return _Method() + + +def test_transient_actor_unavailable_maps_to_503( + fake_resolver: FakeResolver, +) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-unavail0001"] = _UnavailableHandle() + + response = client.get(f"{BASE}/sandboxes/sb-unavail0001") + + assert response.status_code == 503, response.text + assert response.json()["error"]["code"] == "sandbox_unavailable" + + +class _DeadHandle: + """Mimics a handle whose actor has genuinely died (``ActorDiedError``, a + ``RayActorError`` subclass): the app maps it to 404 so a client_token + retry recreates it.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + async def _raise(): + ray_actor_error = type("RayActorError", (Exception,), {}) + exc_type = type("ActorDiedError", (ray_actor_error,), {}) + raise exc_type("actor has died") + + return asyncio.get_running_loop().create_task(_raise()) + + return _Method() + + +def test_dead_actor_maps_to_404(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-deadone0001"] = _DeadHandle() + + response = client.get(f"{BASE}/sandboxes/sb-deadone0001") + + assert response.status_code == 404, response.text + assert response.json()["error"]["code"] == "sandbox_not_found" + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_http_integration.py b/python/ray/experimental/sandbox/http/tests/test_http_integration.py new file mode 100644 index 000000000000..bfe313f94ef8 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_integration.py @@ -0,0 +1,172 @@ +"""End-to-end test of the HTTP API against a real Serve deployment. + +Requires TEST_SANDBOX=1, a local Ray, and runsc (the session fixture in +conftest.py downloads one on Linux). Runs in the Buildkite sandbox job, which +is privileged and sets TEST_SANDBOX=1. +""" + +import os +import shutil +import sys +import time + +import pytest + + +def _sandbox_test_enabled() -> bool: + try: + from ray._private.test_utils import sandbox_test_enabled + except ImportError: + return os.environ.get("TEST_SANDBOX") == "1" + return sandbox_test_enabled() + + +pytestmark = pytest.mark.skipif( + not _sandbox_test_enabled(), + reason="Sandbox tests are only run when TEST_SANDBOX=1", +) + +_TOKEN = "integration-test-token" +_IMAGE = "busybox:latest" + + +@pytest.fixture(scope="module") +def api_base_url(): + if not shutil.which("runsc"): + pytest.skip("runsc is not on PATH") + + # Ray Serve is stripped from the core sandbox CI image (the sandbox + # test job installs with --install-mask all-ray-libraries): the module + # stays importable but loses serve.run, so this end-to-end test only + # runs where Serve is fully installed, e.g. a dev container. + serve = pytest.importorskip("ray.serve") + if not hasattr(serve, "run"): + pytest.skip("ray.serve is present but not fully installed") + + os.environ["RAY_SANDBOX_API_TOKEN"] = _TOKEN + + import ray + from ray.experimental.sandbox.http.app import build_app + + ray.init() + serve.run(build_app({}), name="sandbox-api-integration") + try: + yield "http://127.0.0.1:8000/api/v1" + finally: + serve.shutdown() + ray.shutdown() + + +def _wait_for(fetch, accept, timeout: float = 300.0): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + last = fetch() + if accept(last): + return last + time.sleep(1) + raise AssertionError(f"timed out waiting; last state: {last}") + + +def test_auth_is_enforced(api_base_url): + import httpx + + with httpx.Client(timeout=30) as anonymous: + assert anonymous.get(f"{api_base_url}/health").status_code == 200 + response = anonymous.get(f"{api_base_url}/sandboxes") + assert response.status_code == 401 + assert response.json()["error"]["code"] == "unauthorized" + + +def test_full_sandbox_lifecycle(api_base_url): + import httpx + + headers = {"Authorization": f"Bearer {_TOKEN}"} + with httpx.Client(headers=headers, timeout=60) as client: + response = client.post( + f"{api_base_url}/sandboxes", + json={ + "image": _IMAGE, + "readonly": False, + "network": "none", + "ttl_seconds": 600, + "labels": {"suite": "integration"}, + }, + ) + assert response.status_code == 202, response.text + sandbox_id = response.json()["sandbox_id"] + + info = _wait_for( + lambda: client.get( + f"{api_base_url}/sandboxes/{sandbox_id}", + params={"wait_seconds": 10}, + ).json(), + lambda i: i["status"] in ("running", "error"), + ) + assert info["status"] == "running", info + + # Exec: submit then poll. + response = client.post( + f"{api_base_url}/sandboxes/{sandbox_id}/execs", + json={ + "command": ["sh", "-c", "echo hello-from-sandbox && exit 4"], + "timeout_seconds": 60, + }, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + result = _wait_for( + lambda: client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/execs/{exec_id}", + params={"wait_seconds": 10}, + ).json(), + lambda r: r["status"] != "running", + timeout=120, + ) + assert result["status"] == "completed", result + assert result["exit_code"] == 4 + assert "hello-from-sandbox" in result["stdout"] + + # Files: round-trip through the API and verify inside the sandbox. + payload = b"file-payload-123" + response = client.put( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/in.txt"}, + content=payload, + ) + assert response.status_code == 204, response.text + response = client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/in.txt"}, + ) + assert response.status_code == 200 + assert response.content == payload + + response = client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/does-not-exist"}, + ) + assert response.status_code == 404 + + # Listing sees it. + response = client.get( + f"{api_base_url}/sandboxes", params=[("label", "suite=integration")] + ) + assert sandbox_id in {s["sandbox_id"] for s in response.json()["sandboxes"]} + + # Delete is effective and idempotent. + assert ( + client.delete(f"{api_base_url}/sandboxes/{sandbox_id}").status_code == 200 + ) + _wait_for( + lambda: client.get(f"{api_base_url}/sandboxes/{sandbox_id}").status_code, + lambda code: code == 404, + timeout=60, + ) + assert ( + client.delete(f"{api_base_url}/sandboxes/{sandbox_id}").status_code == 200 + ) + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", "-s", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_http_schemas.py b/python/ray/experimental/sandbox/http/tests/test_http_schemas.py new file mode 100644 index 000000000000..6f87cc29e608 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_schemas.py @@ -0,0 +1,147 @@ +"""Contract snapshot for the Ray Sandbox HTTP API. + +Pins the v1 surface (paths, methods, and the fields of the wire models) so a +change to it is a deliberate act — clients like Harbor's ``ray-sandbox`` +environment are written against exactly this shape. If this test fails, you +are changing the API contract: update it and the consumers together. +""" + +import sys + +import pytest + +from ray.experimental.sandbox.http.app import create_app +from ray.experimental.sandbox.http.schemas import SandboxAPISettings +from ray.experimental.sandbox.http.tests.conftest import FakeResolver + +EXPECTED_OPERATIONS = { + ("/api/v1/health", "get"), + ("/api/v1/sandboxes", "get"), + ("/api/v1/sandboxes", "post"), + ("/api/v1/sandboxes/{sandbox_id}", "get"), + ("/api/v1/sandboxes/{sandbox_id}", "delete"), + ("/api/v1/sandboxes/{sandbox_id}/execs", "post"), + ("/api/v1/sandboxes/{sandbox_id}/execs/{exec_id}", "get"), + ("/api/v1/sandboxes/{sandbox_id}/files", "put"), + ("/api/v1/sandboxes/{sandbox_id}/files", "get"), +} + +EXPECTED_MODEL_FIELDS = { + "CreateSandboxRequest": { + "image", + "env", + "workdir", + "ttl_seconds", + "network", + "rootless", + "readonly", + "resources", + "labels", + "capabilities", + "image_pull_timeout_seconds", + "start_timeout_seconds", + "client_token", + "dns", + "shell", + }, + "ResourceSpec": { + "cpu_request", + "cpu_limit", + "memory_request_mb", + "memory_limit_mb", + "custom", + }, + "SandboxInfo": { + "sandbox_id", + "status", + "image", + "created_at", + "ttl_seconds", + "expires_at", + "network", + "labels", + "error", + }, + "StartExecRequest": { + "command", + "cwd", + "env", + "timeout_seconds", + "shell", + "user", + }, + "ExecStarted": {"exec_id", "status"}, + "ExecInfo": { + "exec_id", + "status", + "exit_code", + "stdout", + "stderr", + "stdout_truncated", + "stderr_truncated", + "duration_seconds", + "error", + }, +} + +_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head", "options"} + + +def _openapi() -> dict: + app = create_app(SandboxAPISettings(), handle_resolver=FakeResolver()) + return app.openapi() + + +def test_v1_operations_are_exactly_the_contract() -> None: + schema = _openapi() + operations = { + (path, method) + for path, item in schema["paths"].items() + for method in item + if method in _HTTP_METHODS + } + assert operations == EXPECTED_OPERATIONS + + +def test_wire_models_carry_exactly_the_contract_fields() -> None: + schema = _openapi() + components = schema["components"]["schemas"] + for model, expected_fields in EXPECTED_MODEL_FIELDS.items(): + assert model in components, f"model {model} missing from the OpenAPI schema" + actual = set(components[model].get("properties", {})) + assert actual == expected_fields, f"{model} fields drifted" + + +def test_sandbox_status_enum_is_stable() -> None: + schema = _openapi() + components = schema["components"]["schemas"] + status = components["SandboxInfo"]["properties"]["status"] + # pydantic may inline or $ref the literal; normalize. + enum = status.get("enum") or components.get( + status.get("$ref", "").rsplit("/", 1)[-1], {} + ).get("enum") + assert set(enum) == { + "pending", + "pulling", + "starting", + "running", + "error", + "terminated", + } + + +def test_network_modes_track_the_core_config() -> None: + """Mode validation is delegated to the core config's list, so a new Ray + mode is accepted here without an API change.""" + from ray.experimental.sandbox.config import VALID_NETWORK_MODES + from ray.experimental.sandbox.http.schemas import CreateSandboxRequest + + for mode in VALID_NETWORK_MODES: + request = CreateSandboxRequest(image="python:3.12", network=mode) + assert request.network == mode + with pytest.raises(ValueError, match="network must be one of"): + CreateSandboxRequest(image="python:3.12", network="bridge") + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py b/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py new file mode 100644 index 000000000000..5998b85782c0 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py @@ -0,0 +1,462 @@ +"""Unit tests for SandboxHost against a fake runtime (no Ray, no runsc).""" + +import asyncio +import sys +import threading + +import pytest + +from ray.experimental.sandbox.exceptions import ( + SandboxCreationError, + SandboxExecError, + SandboxTimeoutError, +) +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import DOCKER_DEFAULT_CAPABILITIES +from ray.experimental.sandbox.http.tests.conftest import ( + FakeExecResult, + FakeSandboxRuntime, +) + + +def _make_host( + runtime: FakeSandboxRuntime, + *, + spec_overrides=None, + settings_overrides=None, +) -> SandboxHost: + spec = { + "image": "python:3.12-slim", + "env": {"TASK_VAR": "1"}, + "workdir": "/", + "ttl_seconds": 3600, + "network": "none", + "rootless": True, + "readonly": False, + "capabilities": list(DOCKER_DEFAULT_CAPABILITIES), + "cpu_limit": 2.0, + "memory_limit_mb": 1024, + "image_pull_timeout_seconds": 300.0, + "start_timeout_seconds": 45.0, + "labels": {"harbor-session-id": "abc"}, + } + spec.update(spec_overrides or {}) + settings = {"max_output_bytes": 10 * 1024 * 1024, "max_exec_history": 256} + settings.update(settings_overrides or {}) + return SandboxHost( + sandbox_id="sb-test00000001", + spec=spec, + settings=settings, + runtime_factory=lambda: runtime, + ) + + +# ---------------------------------------------------------------------- +# Boot +# ---------------------------------------------------------------------- + + +def test_boot_reaches_running_with_expected_create_kwargs() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "running" + assert info["error"] is None + assert info["image"] == "python:3.12-slim" + assert info["labels"] == {"harbor-session-id": "abc"} + assert info["ttl_seconds"] == 3600 + assert info["expires_at"] is not None + + assert runtime.pull_calls == [ + {"image": "python:3.12-slim", "timeout_seconds": 300.0} + ] + (create_call,) = runtime.create_calls + assert create_call["image"] == "python:3.12-slim" + assert create_call["cpu"] == 2.0 + assert create_call["memory"] == "1024Mi" + assert create_call["env"] == {"TASK_VAR": "1"} + assert create_call["workdir"] == "/" + # The API owns the TTL; the upstream runtime must not double-manage it. + assert create_call["ttl_seconds"] is None + # Writability follows the runtime's (readonly, workdir) contract; no + # extra knobs are forwarded. + assert "mount_workdir" not in create_call + assert create_call["timeout_seconds"] == 45.0 + assert create_call["rootless"] is True + assert create_call["network"] == "none" + assert create_call["readonly"] is False + assert create_call["capabilities"] == list(DOCKER_DEFAULT_CAPABILITIES) + assert "_oci_spec_transform_fn" not in create_call + + +def test_boot_failure_is_pollable_not_raised() -> None: + runtime = FakeSandboxRuntime() + runtime.pull_error = SandboxCreationError("registry says no") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert "registry says no" in info["error"] + # Nothing was created, so nothing to delete. + assert runtime.deleted == [] + + +def test_boot_create_failure_cleans_up_nothing_and_reports() -> None: + runtime = FakeSandboxRuntime() + runtime.create_error = SandboxCreationError("runsc not found") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert "runsc not found" in info["error"] + + +def test_boot_is_idempotent() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> None: + await asyncio.gather(host.boot(), host.boot()) + await host.boot() + + asyncio.run(scenario()) + assert len(runtime.pull_calls) == 1 + assert len(runtime.create_calls) == 1 + + +def test_describe_long_poll_wakes_on_status_change() -> None: + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + host = _make_host(runtime) + + async def scenario() -> dict: + boot_task = asyncio.create_task(host.boot()) + # Wait until the pull is actually in flight. + while not runtime.pull_calls: + await asyncio.sleep(0.01) + assert (await host.describe())["status"] == "pulling" + + waiter = asyncio.create_task(host.describe(wait_seconds=10.0)) + await asyncio.sleep(0.05) + runtime.pull_gate.set() + info = await asyncio.wait_for(waiter, timeout=5.0) + await boot_task + return info + + info = asyncio.run(scenario()) + # The long-poll woke on the pulling->starting transition (or later). + assert info["status"] in ("starting", "running") + + +# ---------------------------------------------------------------------- +# Capabilities and network passthrough (behavior itself is covered by the +# core sandbox tests; the host only forwards SandboxConfig fields) +# ---------------------------------------------------------------------- + + +def test_network_mode_passed_through_natively() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"network": "sandbox"}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["network"] == "sandbox" + + +def test_explicit_capabilities_passed_through() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"capabilities": ["CAP_CHOWN"]}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["capabilities"] == ["CAP_CHOWN"] + + +def test_missing_capabilities_default_to_docker_set() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"capabilities": None}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["capabilities"] == list(DOCKER_DEFAULT_CAPABILITIES) + + +# ---------------------------------------------------------------------- +# Exec jobs +# ---------------------------------------------------------------------- + + +def test_exec_job_completes_with_passthrough_args() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [ + FakeExecResult(exit_code=7, stdout="out", stderr="err", duration_seconds=1.5) + ] + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec( + ["bash", "-c", "exit 7"], + cwd="/app", + env={"K": "V"}, + timeout_seconds=30.0, + ) + assert started["status"] == "running" + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "completed" + assert info["exit_code"] == 7 + assert info["stdout"] == "out" + assert info["stderr"] == "err" + assert info["duration_seconds"] == 1.5 + assert info["stdout_truncated"] is False + + exec_call = runtime.exec_calls[-1] + assert exec_call["command"] == ["bash", "-c", "exit 7"] + assert exec_call["cwd"] == "/app" + assert exec_call["env"] == {"K": "V"} + assert exec_call["timeout"] == 30.0 + + +def test_exec_timeout_maps_to_timeout_status() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_error = SandboxTimeoutError("too slow") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("sleep 100", timeout_seconds=1.0) + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "timeout" + assert "timed out after 1.0 seconds" in info["error"] + assert info["exit_code"] is None + + +@pytest.mark.parametrize( + "error", [SandboxExecError("backend broke"), ValueError("surprise")] +) +def test_exec_errors_map_to_error_status(error: Exception) -> None: + runtime = FakeSandboxRuntime() + runtime.exec_error = error + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("true") + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert str(error) in info["error"] + + +def test_exec_output_truncated_with_marker() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [FakeExecResult(stdout="x" * 100, stderr="y")] + host = _make_host(runtime, settings_overrides={"max_output_bytes": 10}) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("true") + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["stdout_truncated"] is True + assert info["stdout"].startswith("x" * 10) + assert "[truncated by ray-sandbox" in info["stdout"] + assert info["stderr_truncated"] is False + assert info["stderr"] == "y" + + +def test_exec_rejected_while_not_running() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + return await host.start_exec("true") + + result = asyncio.run(scenario()) + assert result["error_code"] == "conflict" + assert "pending" in result["message"] + + +def test_get_exec_unknown_id() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.get_exec("ex-doesnotexist") + + result = asyncio.run(scenario()) + assert result["error_code"] == "exec_not_found" + + +def test_exec_history_evicts_oldest_finished() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, settings_overrides={"max_exec_history": 2}) + + async def scenario() -> tuple: + await host.boot() + ids = [] + for _ in range(3): + started = await host.start_exec("true") + await host.get_exec(started["exec_id"], wait_seconds=10.0) + ids.append(started["exec_id"]) + first = await host.get_exec(ids[0]) + last = await host.get_exec(ids[-1]) + return first, last + + first, last = asyncio.run(scenario()) + assert first["error_code"] == "exec_not_found" + assert last["status"] == "completed" + + +# ---------------------------------------------------------------------- +# Files +# ---------------------------------------------------------------------- + + +def test_file_roundtrip_and_not_found() -> None: + runtime = FakeSandboxRuntime() + runtime.readable_files["/data/in.txt"] = b"hello" + host = _make_host(runtime) + + async def scenario() -> tuple: + await host.boot() + wrote = await host.write_file("/data/out.txt", b"payload") + read = await host.read_file("/data/in.txt") + missing = await host.read_file("/data/nope.txt") + return wrote, read, missing + + wrote, read, missing = asyncio.run(scenario()) + assert wrote == {"ok": True} + assert runtime.written_files["/data/out.txt"] == b"payload" + assert read["ok"] is True + assert read["content"] == b"hello" + assert missing["error_code"] == "file_not_found" + + +def test_file_ops_conflict_before_running() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> tuple: + return ( + await host.write_file("/x", b"1"), + await host.read_file("/x"), + ) + + wrote, read = asyncio.run(scenario()) + assert wrote["error_code"] == "conflict" + assert read["error_code"] == "conflict" + + +# ---------------------------------------------------------------------- +# TTL and termination +# ---------------------------------------------------------------------- + + +def test_ttl_deletes_sandbox_and_marks_terminated() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"ttl_seconds": 0.05}) + + async def scenario() -> dict: + await host.boot() + await asyncio.sleep(0.3) + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "terminated" + assert runtime.deleted == [runtime.instance_id] + + +def test_terminate_cancels_running_exec_and_deletes() -> None: + runtime = FakeSandboxRuntime() + exec_gate = threading.Event() + original_exec = runtime.exec + + def blocking_exec(*args, **kwargs): + exec_gate.wait(timeout=30) + return original_exec(*args, **kwargs) + + runtime.exec = blocking_exec + host = _make_host(runtime) + + async def scenario() -> tuple: + await host.boot() + started = await host.start_exec("sleep forever") + await asyncio.sleep(0.05) + try: + result = await host.terminate() + info = await host.describe() + exec_info = await host.get_exec(started["exec_id"]) + finally: + exec_gate.set() + return result, info, exec_info + + result, info, exec_info = asyncio.run(scenario()) + assert result == {"ok": True} + assert info["status"] == "terminated" + assert runtime.deleted == [runtime.instance_id] + assert exec_info["status"] == "error" + assert "terminated" in exec_info["error"] + + +def test_terminate_is_idempotent() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> None: + await host.boot() + await host.terminate() + await host.terminate() + + asyncio.run(scenario()) + assert runtime.deleted == [runtime.instance_id] + + +def test_shell_passthrough() -> None: + """A sandbox-level shell reaches runtime.create only when set, and a + per-exec shell reaches exec_async.""" + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"shell": "/bin/sh"}) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("echo hi", shell="/bin/dash") + return await host.get_exec(started["exec_id"], wait_seconds=5) + + result = asyncio.run(scenario()) + assert result["status"] == "completed" + assert runtime.create_calls[0]["shell"] == "/bin/sh" + assert runtime.exec_calls[0]["shell"] == "/bin/dash" + + +def test_no_shell_in_spec_keeps_the_runtime_default() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + asyncio.run(host.boot()) + # Omitted, not None: SandboxConfig.shell keeps its /bin/bash default. + assert "shell" not in runtime.create_calls[0] + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/runtime.py b/python/ray/experimental/sandbox/runtime.py index 2e24dca6ad00..da0f21b52f05 100644 --- a/python/ray/experimental/sandbox/runtime.py +++ b/python/ray/experimental/sandbox/runtime.py @@ -141,6 +141,7 @@ def exec( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a command inside the specified sandbox. @@ -152,6 +153,8 @@ def exec( env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). + user: Optional user to run as: a numeric uid, "uid:gid", or a + name from the image's /etc/passwd (default: the image user). Returns: ExecResult containing exit code, stdout, and stderr. @@ -163,6 +166,7 @@ def exec( cwd=cwd, env=env, shell=shell, + user=user, ) async def exec_async( @@ -173,6 +177,7 @@ async def exec_async( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a command inside the specified sandbox asynchronously. @@ -184,6 +189,8 @@ async def exec_async( env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). + user: Optional user to run as: a numeric uid, "uid:gid", or a + name from the image's /etc/passwd (default: the image user). Returns: ExecResult containing exit code, stdout, and stderr. @@ -196,6 +203,7 @@ async def exec_async( cwd=cwd, env=env, shell=shell, + user=user, ) def upload_file(self, instance_id: str, local_path: str, remote_path: str) -> None: @@ -228,7 +236,11 @@ def download_file( f.write(content) def write_file( - self, instance_id: str, path: str, content: Union[str, bytes] + self, + instance_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: """Write string or binary content directly to a file inside the sandbox. @@ -236,8 +248,9 @@ def write_file( instance_id: Unique identifier of the sandbox instance. path: Destination file path inside the sandbox. content: String or binary content to write into the file. + append: Append to the file instead of truncating it. """ - self._backend.write_file(instance_id, path, content) + self._backend.write_file(instance_id, path, content, append=append) def read_file(self, instance_id: str, path: str) -> bytes: """Read binary content from a file inside the sandbox. diff --git a/python/ray/experimental/sandbox/tests/test_gvisor_backend.py b/python/ray/experimental/sandbox/tests/test_gvisor_backend.py index 8684755f4e6a..b5fd80810691 100644 --- a/python/ray/experimental/sandbox/tests/test_gvisor_backend.py +++ b/python/ray/experimental/sandbox/tests/test_gvisor_backend.py @@ -11,6 +11,7 @@ from ray.experimental.sandbox.config import GVisorSandboxConfig from ray.experimental.sandbox.exceptions import ( SandboxCreationError, + SandboxExecError, SandboxNotFoundError, ) from ray.experimental.sandbox.runtime import SandboxRuntime @@ -305,5 +306,25 @@ def test_image_workdir_sets_cwd_without_becoming_writable(): runtime.delete(instance_id) +def test_resolve_exec_user(tmp_path, monkeypatch): + """Numeric users pass through; names resolve via the image's passwd.""" + backend = GVisorSandboxBackend() + img_dir = tmp_path / "img" + (img_dir / "rootfs" / "etc").mkdir(parents=True) + (img_dir / "rootfs" / "etc" / "passwd").write_text( + "root:x:0:0:root:/root:/bin/bash\n" + "postfix:x:102:104::/var/spool/postfix:/usr/sbin/nologin\n" + ) + monkeypatch.setattr( + backend._image_manager, "get_image_dir", lambda image: str(img_dir) + ) + + assert backend._resolve_exec_user("1000", "img") == "1000" + assert backend._resolve_exec_user("1000:1000", "img") == "1000:1000" + assert backend._resolve_exec_user("postfix", "img") == "102:104" + with pytest.raises(SandboxExecError): + backend._resolve_exec_user("nosuch", "img") + + if __name__ == "__main__": sys.exit(pytest.main(["-v", __file__]))