Source code for ray.experimental.sandbox.runtime

import asyncio
import os
import threading
from typing import Callable, Dict, List, Optional, Union

from ray.experimental.sandbox.backend.base import (
    ExecResult,
    SandboxStatus,
)
from ray.experimental.sandbox.backend.gvisor import GVisorSandboxBackend
from ray.experimental.sandbox.config import SandboxConfig
from ray.experimental.sandbox.image_manager import ImageManager
from ray.util.annotations import PublicAPI


[docs] @PublicAPI(stability="alpha") class SandboxRuntime: """Low-level interface for managing local sandbox runtime environments.""" def __init__(self): self._image_manager = ImageManager() self._backend = GVisorSandboxBackend(image_manager=self._image_manager) self._ttl_timers: Dict[str, threading.Timer] = {} @property def image_manager(self) -> ImageManager: """The ImageManager instance used by this runtime.""" return self._image_manager @property def backend(self) -> GVisorSandboxBackend: """The backend instance used by this runtime.""" return self._backend
[docs] def pull_image(self, image: str, timeout_seconds: float = 120.0) -> str: """Download and extract container image into local cache. Args: image: Container image name or tar path. timeout_seconds: Request timeout. Returns: Extracted image directory path. """ return self._image_manager.pull_image(image, timeout_seconds=timeout_seconds)
[docs] def create( self, image: str, cpu: float = 0.0, memory: Union[str, int, float] = 0, env: Optional[Dict[str, str]] = None, workdir: Optional[str] = None, ttl_seconds: Optional[int] = None, timeout_seconds: float = 30.0, rootless: bool = True, network: str = "none", dns: Optional[List[str]] = None, capabilities: Optional[List[str]] = None, readonly: bool = True, _oci_spec_transform_fn: Optional[Callable[[Dict], Optional[Dict]]] = None, _ignore_cgroups: bool = False, **kwargs, ) -> str: """Provision the sandbox instance and return unique instance ID. Args: image: Container image for the sandbox environment. cpu: Number of CPU cores allocated to the sandbox. memory: Amount of memory allocated to the sandbox (e.g. "1Gi", "512Mi"). env: Environment variables to inject into the sandbox. workdir: Working directory for commands; None uses the image's WORKDIR. On a readonly rootfs an explicit workdir is also the sandbox's only writable path; see :class:`~ray.experimental.sandbox.config.SandboxConfig`. ttl_seconds: Optional time-to-live in seconds, wall-clock from creation (not idle time), enforced by this runtime with a daemon timer. None (default) or <= 0 disables it. timeout_seconds: Timeout in seconds for sandbox creation. rootless: If True, run gVisor in rootless mode. network: Network mode ("none", "public", "host", "sandbox"); see :class:`~ray.experimental.sandbox.config.SandboxConfig`. "public" is the recommended internet-access mode. dns: Optional nameserver IPs for the generated /etc/resolv.conf (public resolvers by default for "public"). capabilities: Linux capabilities, written exactly (None keeps the runtime default; ``[]`` means none). Use ``DOCKER_DEFAULT_CAPABILITIES`` for Docker parity. readonly: If True (default), mount container image rootfs in read-only mode such that only ``workdir`` is writable. If False, the entire root filesystem is writable. Writes are isolated within a per-sandbox copy-on-write overlay filesystem, ensuring multiple sandboxes running the same container image do not interfere with each other or modify the base image. _oci_spec_transform_fn: PRIVATE — development/testing only. Called with the fully-built OCI spec dict before it is written; may mutate in place or return a new dict. Must be cloudpickle-serializable. No stability guarantees. Accepts a transform function. _ignore_cgroups: PRIVATE — testing only. If True, passes --ignore-cgroups to runsc. **kwargs: Additional parameters. Returns: A unique string identifier for the created sandbox. """ cfg = SandboxConfig( image=image, cpu=cpu, memory=memory, env=env or {}, workdir=workdir, ttl_seconds=ttl_seconds, timeout_seconds=timeout_seconds, rootless=rootless, network=network, dns=dns, capabilities=capabilities, readonly=readonly, _oci_spec_transform_fn=_oci_spec_transform_fn, _ignore_cgroups=_ignore_cgroups, **kwargs, ) self._image_manager.pull_image(cfg.image, timeout_seconds=cfg.timeout_seconds) instance_id = self._backend.create_sandbox(cfg) if cfg.ttl_seconds is not None and cfg.ttl_seconds > 0: timer = threading.Timer(cfg.ttl_seconds, self._expire, args=(instance_id,)) timer.daemon = True self._ttl_timers[instance_id] = timer timer.start() return instance_id
def _expire(self, instance_id: str) -> None: """TTL callback: best-effort delete (the sandbox may already be gone).""" try: self.delete(instance_id) except Exception: pass
[docs] 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, ) -> ExecResult: """Execute a command inside the specified sandbox. Args: instance_id: Unique identifier of the sandbox instance. command: Command to execute, either as a string or a list of arguments. timeout: Maximum execution time in seconds. cwd: Working directory inside the sandbox for command execution. env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). Returns: ExecResult containing exit code, stdout, and stderr. """ return self._backend.exec_command( instance_id, command, timeout=timeout, cwd=cwd, env=env, shell=shell, )
[docs] 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, ) -> ExecResult: """Execute a command inside the specified sandbox asynchronously. Args: instance_id: Unique identifier of the sandbox instance. command: Command to execute, either as a string or a list of arguments. timeout: Maximum execution time in seconds. cwd: Working directory inside the sandbox for command execution. env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). Returns: ExecResult containing exit code, stdout, and stderr. """ return await asyncio.to_thread( self.exec, instance_id, command, timeout=timeout, cwd=cwd, env=env, shell=shell, )
[docs] def upload_file(self, instance_id: str, local_path: str, remote_path: str) -> None: """Copy local file into the sandbox. Args: instance_id: Unique identifier of the sandbox instance. local_path: Path to the source file on the local filesystem. remote_path: Destination path inside the sandbox. """ with open(local_path, "rb") as f: content = f.read() self._backend.write_file(instance_id, remote_path, content)
[docs] def download_file( self, instance_id: str, remote_path: str, local_path: str ) -> None: """Copy file from the sandbox to local. Args: instance_id: Unique identifier of the sandbox instance. remote_path: Path to the source file inside the sandbox. local_path: Destination path on the local filesystem. """ content = self._backend.read_file(instance_id, remote_path) local_dir = os.path.dirname(os.path.abspath(local_path)) if local_dir: os.makedirs(local_dir, exist_ok=True) with open(local_path, "wb") as f: f.write(content)
[docs] def write_file( self, instance_id: str, path: str, content: Union[str, bytes] ) -> None: """Write string or binary content directly to a file inside the sandbox. Args: 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. """ self._backend.write_file(instance_id, path, content)
[docs] def read_file(self, instance_id: str, path: str) -> bytes: """Read binary content from a file inside the sandbox. Args: instance_id: Unique identifier of the sandbox instance. path: Path to the file inside the sandbox to read. Returns: File content as bytes. """ return self._backend.read_file(instance_id, path)
[docs] def get_status(self, instance_id: str) -> SandboxStatus: """Query operational status of the sandbox. Args: instance_id: Unique identifier of the sandbox instance. Returns: SandboxStatus of the sandbox instance. """ return self._backend.get_status(instance_id)
[docs] def delete(self, instance_id: str) -> None: """Clean up and terminate the sandbox instance. Args: instance_id: Unique identifier of the sandbox instance. """ timer = self._ttl_timers.pop(instance_id, None) if timer is not None: timer.cancel() self._backend.delete_sandbox(instance_id)
[docs] def terminate(self, instance_id: str) -> None: """Clean up and terminate the sandbox instance. Args: instance_id: Unique identifier of the sandbox instance. """ self.delete(instance_id)
[docs] async def delete_async(self, instance_id: str) -> None: """Clean up and terminate the sandbox instance asynchronously. Args: instance_id: Unique identifier of the sandbox instance. """ await asyncio.to_thread(self.delete, instance_id)