perf(sandbox): delete containers through worker rocklet

rock/sandbox/operator/ray.py

@@ -1,16 +1,21 @@
 import json
+import posixpath
+import shlex
 
 import ray
 
 from rock.actions.sandbox.response import State
 from rock.actions.sandbox.sandbox_info import SandboxInfo
 from rock.admin.core.ray_service import RayService
+from rock.admin.proto.request import SandboxCommand as Command
 from rock.common.constants import StopReason
 from rock.config import RuntimeConfig
 from rock.deployments.config import DockerDeploymentConfig
+from rock.deployments.constants import Port
 from rock.deployments.docker import DockerDeployment
 from rock.logger import init_logger
 from rock.sandbox.operator.abstract import AbstractOperator
+from rock.sandbox.remote_sandbox import RemoteSandboxRuntime
 from rock.sandbox.sandbox_actor import SandboxActor
 from rock.sandbox.utils.rocklet_probe import check_alive_status, get_remote_status
 from rock.sdk.common.exceptions import BadRequestRockError
@@ -139,35 +144,40 @@ async def stop(self, sandbox_id: str, reason: StopReason = StopReason.MANUAL) ->
             return True
 
     async def delete(self, config: DockerDeploymentConfig, host_ip: str | None = None) -> bool:
-        async with self._ray_service.get_ray_rwlock().read_lock():
-            sandbox_id = config.container_name
-            actor_name = self._get_actor_name(sandbox_id)
-
-            try:
-                existing_actor = await self._ray_service.async_ray_get_actor(actor_name)
-                ray.kill(existing_actor)
-            except Exception:
-                logger.info(f"Actor {actor_name} already gone, proceeding with delete")
-
-            if not host_ip:
-                logger.warning(
-                    f"delete for {sandbox_id} called without host_ip; new actor "
-                    f"may be scheduled on a node that does not own the container"
-                )
-            # Cleanup must remain schedulable when the worker's logical
-            # sandbox resources are exhausted. Ray still starts a real worker
-            # process, but it does not reserve CPU, memory, or disk capacity
-            # for this short-lived actor.
-            config.cpus = 0
-            config.memory = "0"
-            config.disk = None
-            sandbox_actor: SandboxActor = await self.create_actor(config, pin_to_host_ip=host_ip)
-            try:
-                await self._ray_service.async_ray_get(sandbox_actor.delete.remote())
-                logger.info(f"sandbox {sandbox_id} deleted on host_ip={host_ip}")
-                return True
-            finally:
-                ray.kill(sandbox_actor)
+        sandbox_id = config.container_name
+        if not host_ip:
+            raise ValueError(f"delete for {sandbox_id} requires the worker host_ip")
+
+        # Use the worker's host-level rocklet, as scheduler cleanup tasks do,
+        # instead of starting a short-lived Ray actor solely to run docker rm.
+        runtime = RemoteSandboxRuntime(host=host_ip, port=Port.PROXY.value)
+        delete_command = f"docker rm -f -v {shlex.quote(sandbox_id)}"
+        if config.use_kata_runtime:
+            disk_path = posixpath.join(config.kata_disk_base_path, f"{sandbox_id}.img")
+            delete_command = (
+                f"{delete_command}; docker_status=$?; "
+                f"rm -f -- {shlex.quote(disk_path)}; kata_status=$?; "
+                '[ "$docker_status" -eq 0 ] && [ "$kata_status" -eq 0 ]'
+            )
+
+        result = await runtime.execute(
+            Command(
+                command=delete_command,
+                timeout=10,
+                shell=True,
+                check=False,
+                sandbox_id=sandbox_id,
+            )
+        )
+        if result.exit_code != 0:
+            # Deletion is idempotent: a missing container is already clean.
+            logger.warning(
+                f"worker cleanup for sandbox {sandbox_id} on host_ip={host_ip} "
+                f"returned exit_code={result.exit_code}: {result.stderr}"
+            )
+
+        logger.info(f"sandbox {sandbox_id} deleted through rocklet on host_ip={host_ip}")
+        return True
 
     async def restart(self, config: DockerDeploymentConfig, host_ip: str | None = None) -> SandboxInfo:
         """Restart an existing sandbox using docker start (container is preserved).

tests/unit/sandbox/operator/test_ray_operator_delete.py

@@ -2,6 +2,7 @@
 
 import pytest
 
+from rock.actions import CommandResponse
 from rock.admin.core.ray_service import RayService
 from rock.config import RayConfig, RuntimeConfig
 from rock.deployments.config import DockerDeploymentConfig
@@ -16,58 +17,91 @@ def _make_operator() -> tuple[RayOperator, RayService]:
 
 
 @pytest.mark.asyncio
-async def test_delete_actor_drops_sandbox_resources():
+async def test_delete_uses_worker_rocklet_without_creating_actor():
     operator, ray_service = _make_operator()
-    actor = MagicMock()
-    delete_ref = object()
-    actor.delete.remote.return_value = delete_ref
-    operator.create_actor = AsyncMock(return_value=actor)
-    ray_service.async_ray_get_actor = AsyncMock(side_effect=ValueError("actor not found"))
-    ray_service.async_ray_get = AsyncMock(return_value=None)
+    operator.create_actor = AsyncMock()
+    ray_service.async_ray_get_actor = AsyncMock()
+    ray_service.get_ray_rwlock = MagicMock(side_effect=AssertionError("delete must not acquire the Ray lock"))
+    runtime = MagicMock()
+    runtime.execute = AsyncMock(return_value=CommandResponse(stdout="sb-1\n", stderr="", exit_code=0))
     config = DockerDeploymentConfig(
         container_name="sb-1",
         cpus=4,
         memory="8g",
         disk="128g",
     )
 
-    with patch("rock.sandbox.operator.ray.ray.kill") as kill:
+    with (
+        patch("rock.sandbox.operator.ray.RemoteSandboxRuntime", return_value=runtime) as runtime_cls,
+        patch("rock.sandbox.operator.ray.ray.kill") as kill,
+    ):
         result = await operator.delete(config, host_ip="10.0.0.1")
 
     assert result is True
-    assert config.cpus == 0
-    assert config.memory == "0"
-    assert config.disk is None
-    operator._disk_scheduling_enabled = True
-    actor_options = operator._generate_actor_options(config, pin_to_host_ip="10.0.0.1")
-    assert actor_options["num_cpus"] == 0
-    assert actor_options["memory"] == 0
-    assert actor_options["resources"] == {"node:10.0.0.1": 0.001}
-    operator.create_actor.assert_awaited_once_with(config, pin_to_host_ip="10.0.0.1")
-    ray_service.async_ray_get.assert_awaited_once_with(delete_ref)
-    kill.assert_called_once_with(actor)
+    assert config.cpus == 4
+    assert config.memory == "8g"
+    assert config.disk == "128g"
+    operator.create_actor.assert_not_awaited()
+    ray_service.get_ray_rwlock.assert_not_called()
+    ray_service.async_ray_get_actor.assert_not_awaited()
+    kill.assert_not_called()
+    runtime_cls.assert_called_once_with(host="10.0.0.1", port=22555)
+    runtime.execute.assert_awaited_once()
+    command = runtime.execute.await_args.args[0]
+    assert command.command == "docker rm -f -v sb-1"
+    assert command.timeout == 10
+    assert command.shell is True
+    assert command.check is False
+    assert command.sandbox_id == "sb-1"
 
 
 @pytest.mark.asyncio
-async def test_delete_actor_timeout_propagates_and_kills_pending_actor():
-    operator, ray_service = _make_operator()
-    actor = MagicMock()
-    actor.delete.remote.return_value = object()
-    operator.create_actor = AsyncMock(return_value=actor)
-    ray_service.async_ray_get_actor = AsyncMock(side_effect=ValueError("actor not found"))
-    ray_service.async_ray_get = AsyncMock(side_effect=Exception("ray get timed out"))
+async def test_delete_rocklet_failure_propagates():
+    operator, _ = _make_operator()
+    runtime = MagicMock()
+    runtime.execute = AsyncMock(side_effect=Exception("rocklet timed out"))
     config = DockerDeploymentConfig(container_name="sb-1", disk="128g")
 
     with (
-        patch("rock.sandbox.operator.ray.ray.kill") as kill,
-        pytest.raises(Exception, match="ray get timed out"),
+        patch("rock.sandbox.operator.ray.RemoteSandboxRuntime", return_value=runtime),
+        pytest.raises(Exception, match="rocklet timed out"),
     ):
         await operator.delete(config, host_ip="10.0.0.1")
 
-    assert config.disk is None
-    ray_service.async_ray_get.assert_awaited_once()
-    assert ray_service.async_ray_get.await_args.kwargs == {}
-    kill.assert_called_once_with(actor)
+    runtime.execute.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_delete_kata_disk_through_worker_rocklet():
+    operator, _ = _make_operator()
+    runtime = MagicMock()
+    runtime.execute = AsyncMock(return_value=CommandResponse(stdout="sb-1\n", stderr="", exit_code=0))
+    config = DockerDeploymentConfig(
+        container_name="sb-1",
+        use_kata_runtime=True,
+        kata_disk_base_path="/data/docker-disk",
+    )
+
+    with patch("rock.sandbox.operator.ray.RemoteSandboxRuntime", return_value=runtime):
+        await operator.delete(config, host_ip="10.0.0.1")
+
+    runtime.execute.assert_awaited_once()
+    command = runtime.execute.await_args.args[0]
+    assert command.command == (
+        "docker rm -f -v sb-1; docker_status=$?; "
+        "rm -f -- /data/docker-disk/sb-1.img; kata_status=$?; "
+        '[ "$docker_status" -eq 0 ] && [ "$kata_status" -eq 0 ]'
+    )
+    assert command.timeout == 10
+    assert command.shell is True
+
+
+@pytest.mark.asyncio
+async def test_delete_requires_worker_host_ip():
+    operator, _ = _make_operator()
+
+    with pytest.raises(ValueError, match="requires the worker host_ip"):
+        await operator.delete(DockerDeploymentConfig(container_name="sb-1"))
 
 
 @pytest.mark.asyncio