import asyncio
import os
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)
@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] = 3600,
timeout_seconds: float = 30.0,
rootless: bool = True,
network: 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: Default working directory inside the sandbox. By default, the
working directory is the only writable path in the sandbox (unless
``readonly=False`` is set). If not provided, the container's WORKDIR is used.
ttl_seconds: Optional automatic cleanup time-to-live in seconds.
timeout_seconds: Timeout in seconds for sandbox creation.
rootless: If True, run gVisor in rootless mode.
network: Network mode for runsc.
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,
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)
return self._backend.create_sandbox(cfg)
[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,
) -> 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.
Returns:
ExecResult containing exit code, stdout, and stderr.
"""
return self._backend.exec_command(
instance_id,
command,
timeout=timeout,
cwd=cwd,
env=env,
)
[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,
) -> 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.
Returns:
ExecResult containing exit code, stdout, and stderr.
"""
return await asyncio.to_thread(
self.exec,
instance_id,
command,
timeout=timeout,
cwd=cwd,
env=env,
)
[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.
"""
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)