Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 44e611d14e | |||
| 4fb0f64249 |
@@ -32,6 +32,7 @@ from .broker import (
|
|||||||
sign_request,
|
sign_request,
|
||||||
verify_request,
|
verify_request,
|
||||||
)
|
)
|
||||||
|
from .docker_broker import DockerBroker, DockerBrokerError
|
||||||
from .service import Orchestrator
|
from .service import Orchestrator
|
||||||
from .control_plane import ControlPlaneServer, dispatch, make_server
|
from .control_plane import ControlPlaneServer, dispatch, make_server
|
||||||
|
|
||||||
@@ -43,6 +44,8 @@ __all__ = [
|
|||||||
"LaunchBroker",
|
"LaunchBroker",
|
||||||
"LaunchRequest",
|
"LaunchRequest",
|
||||||
"StubBroker",
|
"StubBroker",
|
||||||
|
"DockerBroker",
|
||||||
|
"DockerBrokerError",
|
||||||
"sign_request",
|
"sign_request",
|
||||||
"verify_request",
|
"verify_request",
|
||||||
"Orchestrator",
|
"Orchestrator",
|
||||||
|
|||||||
@@ -16,8 +16,9 @@ import secrets
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from .. import log
|
from .. import log
|
||||||
from .broker import StubBroker
|
from .broker import LaunchBroker, StubBroker
|
||||||
from .control_plane import make_server
|
from .control_plane import make_server
|
||||||
|
from .docker_broker import DockerBroker
|
||||||
from .registry import RegistryStore, default_db_path
|
from .registry import RegistryStore, default_db_path
|
||||||
from .service import Orchestrator
|
from .service import Orchestrator
|
||||||
|
|
||||||
@@ -31,16 +32,21 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
"--db", type=Path, default=None,
|
"--db", type=Path, default=None,
|
||||||
help=f"registry DB path (default: {default_db_path()})",
|
help=f"registry DB path (default: {default_db_path()})",
|
||||||
)
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--broker", choices=("stub", "docker"), default="stub",
|
||||||
|
help="launch broker: 'stub' records requests; 'docker' runs containers",
|
||||||
|
)
|
||||||
args = parser.parse_args(argv)
|
args = parser.parse_args(argv)
|
||||||
|
|
||||||
registry = RegistryStore(args.db)
|
registry = RegistryStore(args.db)
|
||||||
registry.migrate()
|
registry.migrate()
|
||||||
|
|
||||||
# Dev-harness wiring: an ephemeral signing secret + a stub broker that
|
# An ephemeral signing secret ties the orchestrator (signer) to its
|
||||||
# records launches instead of starting anything. A real backend broker
|
# broker (verifier). 'stub' records launches instead of starting
|
||||||
# (docker, then firecracker) drops in here later.
|
# anything; 'docker' runs real containers (firecracker drops in later).
|
||||||
secret = secrets.token_bytes(32)
|
secret = secrets.token_bytes(32)
|
||||||
orchestrator = Orchestrator(registry, StubBroker(secret), secret)
|
broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret)
|
||||||
|
orchestrator = Orchestrator(registry, broker, secret)
|
||||||
|
|
||||||
server = make_server(orchestrator, host=args.host, port=args.port)
|
server = make_server(orchestrator, host=args.host, port=args.port)
|
||||||
bound_host, bound_port = server.server_address[0], server.server_address[1]
|
bound_host, bound_port = server.server_address[0], server.server_address[1]
|
||||||
|
|||||||
@@ -0,0 +1,85 @@
|
|||||||
|
"""Docker launch broker (PRD 0070) — the first *real* LaunchBroker.
|
||||||
|
|
||||||
|
On a verified launch request it starts a Docker container; on teardown it
|
||||||
|
removes it. This proves the orchestrator -> backend seam on the cheapest
|
||||||
|
backend (the sidecar bundle is already containers). Only the request's
|
||||||
|
static ids/flags reach `docker`, so nothing free-form crosses the boundary.
|
||||||
|
|
||||||
|
Slice 3 launches a single container from the request's `image_ref`, named
|
||||||
|
after the bottle id and labelled for cleanup. Wiring the full agent +
|
||||||
|
sidecar bundle (networks, mounts, the consolidated sidecar) is a later
|
||||||
|
slice — this is the seam, not the finished launcher.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
from .broker import LaunchBroker, LaunchRequest
|
||||||
|
|
||||||
|
CONTAINER_PREFIX = "bot-bottle-orch-"
|
||||||
|
BOTTLE_ID_LABEL = "bot-bottle-bottle-id"
|
||||||
|
|
||||||
|
|
||||||
|
class DockerBrokerError(Exception):
|
||||||
|
"""A brokered docker launch/teardown failed (non-zero `docker` exit)."""
|
||||||
|
|
||||||
|
|
||||||
|
def container_name(bottle_id: str) -> str:
|
||||||
|
"""Deterministic container name for a bottle."""
|
||||||
|
return f"{CONTAINER_PREFIX}{bottle_id}"
|
||||||
|
|
||||||
|
|
||||||
|
def run_argv(req: LaunchRequest) -> list[str]:
|
||||||
|
"""`docker run` argv for a launch request — built only from its static
|
||||||
|
fields, so it can't be coerced into an arbitrary command."""
|
||||||
|
return [
|
||||||
|
"docker", "run", "--detach",
|
||||||
|
"--name", container_name(req.bottle_id),
|
||||||
|
"--label", f"{BOTTLE_ID_LABEL}={req.bottle_id}",
|
||||||
|
req.image_ref,
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def rm_argv(req: LaunchRequest) -> list[str]:
|
||||||
|
"""`docker rm --force` argv to tear a bottle's container down."""
|
||||||
|
return ["docker", "rm", "--force", container_name(req.bottle_id)]
|
||||||
|
|
||||||
|
|
||||||
|
class DockerBroker(LaunchBroker):
|
||||||
|
"""A `LaunchBroker` that runs / removes a Docker container per bottle.
|
||||||
|
Provenance + schema verification are inherited from `LaunchBroker`."""
|
||||||
|
|
||||||
|
def _docker(self, argv: list[str]) -> subprocess.CompletedProcess[str]:
|
||||||
|
return subprocess.run(
|
||||||
|
argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||||
|
text=True, check=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _launch(self, req: LaunchRequest) -> None:
|
||||||
|
if not req.image_ref:
|
||||||
|
raise DockerBrokerError(f"launch request for {req.bottle_id} has no image_ref")
|
||||||
|
proc = self._docker(run_argv(req))
|
||||||
|
if proc.returncode != 0:
|
||||||
|
raise DockerBrokerError(
|
||||||
|
f"docker run failed for {req.bottle_id}: {proc.stderr.strip()}"
|
||||||
|
)
|
||||||
|
|
||||||
|
def _teardown(self, req: LaunchRequest) -> None:
|
||||||
|
proc = self._docker(rm_argv(req))
|
||||||
|
# Idempotent: an already-absent container is a successful teardown.
|
||||||
|
if proc.returncode != 0 and "No such container" not in proc.stderr:
|
||||||
|
raise DockerBrokerError(
|
||||||
|
f"docker rm failed for {req.bottle_id}: {proc.stderr.strip()}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"DockerBroker",
|
||||||
|
"DockerBrokerError",
|
||||||
|
"container_name",
|
||||||
|
"run_argv",
|
||||||
|
"rm_argv",
|
||||||
|
"CONTAINER_PREFIX",
|
||||||
|
"BOTTLE_ID_LABEL",
|
||||||
|
]
|
||||||
@@ -0,0 +1,56 @@
|
|||||||
|
"""Integration: the Docker launch broker starts and removes a real container.
|
||||||
|
|
||||||
|
Gated on a reachable Docker daemon (skips cleanly otherwise). Uses a tiny
|
||||||
|
image; the container may exit immediately — we only assert it exists after
|
||||||
|
launch and is gone after teardown.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import secrets
|
||||||
|
import subprocess
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from bot_bottle.orchestrator.broker import LaunchRequest, sign_request
|
||||||
|
from bot_bottle.orchestrator.docker_broker import DockerBroker, container_name
|
||||||
|
from tests._docker import skip_unless_docker
|
||||||
|
|
||||||
|
IMAGE = "busybox"
|
||||||
|
|
||||||
|
|
||||||
|
@skip_unless_docker()
|
||||||
|
class TestDockerBrokerIntegration(unittest.TestCase):
|
||||||
|
def setUp(self) -> None:
|
||||||
|
self.secret = secrets.token_bytes(16)
|
||||||
|
self.broker = DockerBroker(self.secret)
|
||||||
|
self.bottle_id = "itest" + secrets.token_hex(4)
|
||||||
|
self.name = container_name(self.bottle_id)
|
||||||
|
self.addCleanup(
|
||||||
|
lambda: subprocess.run(
|
||||||
|
["docker", "rm", "--force", self.name],
|
||||||
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _exists(self) -> bool:
|
||||||
|
proc = subprocess.run(
|
||||||
|
["docker", "ps", "-a", "--filter", f"name=^{self.name}$",
|
||||||
|
"--format", "{{.Names}}"],
|
||||||
|
stdout=subprocess.PIPE, text=True, check=False,
|
||||||
|
)
|
||||||
|
return self.name in proc.stdout.split()
|
||||||
|
|
||||||
|
def _submit(self, req: LaunchRequest) -> None:
|
||||||
|
self.broker.submit(sign_request(req, self.secret))
|
||||||
|
|
||||||
|
def test_launch_creates_then_teardown_removes(self) -> None:
|
||||||
|
self._submit(
|
||||||
|
LaunchRequest(op="launch", bottle_id=self.bottle_id, image_ref=IMAGE)
|
||||||
|
)
|
||||||
|
self.assertTrue(self._exists(), "container should exist after launch")
|
||||||
|
self._submit(LaunchRequest(op="teardown", bottle_id=self.bottle_id))
|
||||||
|
self.assertFalse(self._exists(), "container should be gone after teardown")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -0,0 +1,100 @@
|
|||||||
|
"""Unit tests for the Docker launch broker (PRD 0070). Docker is mocked."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import secrets
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import Mock, patch
|
||||||
|
|
||||||
|
from bot_bottle.orchestrator.broker import (
|
||||||
|
BrokerAuthError,
|
||||||
|
LaunchRequest,
|
||||||
|
sign_request,
|
||||||
|
)
|
||||||
|
from bot_bottle.orchestrator.docker_broker import (
|
||||||
|
BOTTLE_ID_LABEL,
|
||||||
|
DockerBroker,
|
||||||
|
DockerBrokerError,
|
||||||
|
container_name,
|
||||||
|
rm_argv,
|
||||||
|
run_argv,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestArgv(unittest.TestCase):
|
||||||
|
def test_run_argv_uses_only_static_fields(self) -> None:
|
||||||
|
req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
|
||||||
|
argv = run_argv(req)
|
||||||
|
self.assertEqual(["docker", "run"], argv[:2])
|
||||||
|
self.assertIn("--name", argv)
|
||||||
|
self.assertIn(container_name("b1"), argv)
|
||||||
|
self.assertIn(f"{BOTTLE_ID_LABEL}=b1", argv)
|
||||||
|
self.assertEqual("busybox", argv[-1]) # image is the terminal arg
|
||||||
|
|
||||||
|
def test_rm_argv(self) -> None:
|
||||||
|
req = LaunchRequest(op="teardown", bottle_id="b1")
|
||||||
|
self.assertEqual(["docker", "rm", "--force", container_name("b1")], rm_argv(req))
|
||||||
|
|
||||||
|
def test_container_name_is_prefixed(self) -> None:
|
||||||
|
name = container_name("b1")
|
||||||
|
self.assertTrue(name.startswith("bot-bottle-orch-"))
|
||||||
|
self.assertTrue(name.endswith("b1"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestDockerBroker(unittest.TestCase):
|
||||||
|
def setUp(self) -> None:
|
||||||
|
self.secret = secrets.token_bytes(16)
|
||||||
|
self.broker = DockerBroker(self.secret)
|
||||||
|
|
||||||
|
def _submit(self, req: LaunchRequest) -> None:
|
||||||
|
self.broker.submit(sign_request(req, self.secret))
|
||||||
|
|
||||||
|
def test_launch_invokes_docker_run(self) -> None:
|
||||||
|
req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
|
||||||
|
with patch.object(self.broker, "_docker", return_value=Mock(returncode=0, stderr="")) as m:
|
||||||
|
self._submit(req)
|
||||||
|
m.assert_called_once()
|
||||||
|
self.assertEqual(run_argv(req), m.call_args.args[0])
|
||||||
|
|
||||||
|
def test_launch_without_image_raises_and_skips_docker(self) -> None:
|
||||||
|
req = LaunchRequest(op="launch", bottle_id="b1", image_ref="")
|
||||||
|
with patch.object(self.broker, "_docker") as m:
|
||||||
|
with self.assertRaises(DockerBrokerError):
|
||||||
|
self._submit(req)
|
||||||
|
m.assert_not_called()
|
||||||
|
|
||||||
|
def test_launch_docker_failure_raises(self) -> None:
|
||||||
|
req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
|
||||||
|
with patch.object(self.broker, "_docker", return_value=Mock(returncode=1, stderr="boom")):
|
||||||
|
with self.assertRaises(DockerBrokerError):
|
||||||
|
self._submit(req)
|
||||||
|
|
||||||
|
def test_teardown_invokes_docker_rm(self) -> None:
|
||||||
|
req = LaunchRequest(op="teardown", bottle_id="b1")
|
||||||
|
with patch.object(self.broker, "_docker", return_value=Mock(returncode=0, stderr="")) as m:
|
||||||
|
self._submit(req)
|
||||||
|
self.assertEqual(rm_argv(req), m.call_args.args[0])
|
||||||
|
|
||||||
|
def test_teardown_is_idempotent_on_missing_container(self) -> None:
|
||||||
|
req = LaunchRequest(op="teardown", bottle_id="b1")
|
||||||
|
absent = Mock(returncode=1, stderr="Error: No such container: bot-bottle-orch-b1")
|
||||||
|
with patch.object(self.broker, "_docker", return_value=absent):
|
||||||
|
self._submit(req) # must not raise
|
||||||
|
|
||||||
|
def test_teardown_other_failure_raises(self) -> None:
|
||||||
|
req = LaunchRequest(op="teardown", bottle_id="b1")
|
||||||
|
with patch.object(self.broker, "_docker", return_value=Mock(returncode=1, stderr="daemon down")):
|
||||||
|
with self.assertRaises(DockerBrokerError):
|
||||||
|
self._submit(req)
|
||||||
|
|
||||||
|
def test_forged_token_never_touches_docker(self) -> None:
|
||||||
|
req = LaunchRequest(op="launch", bottle_id="b1", image_ref="busybox")
|
||||||
|
forged = sign_request(req, secrets.token_bytes(16))
|
||||||
|
with patch.object(self.broker, "_docker") as m:
|
||||||
|
with self.assertRaises(BrokerAuthError):
|
||||||
|
self.broker.submit(forged)
|
||||||
|
m.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user