refactor(orchestrator): introduce the Orchestrator service ABC (docker impl)

Mirror the Gateway work for the control plane. Add the backend-neutral
`Orchestrator` service ABC in orchestrator/lifecycle.py — build the image/rootfs,
`ensure_running` (start + health-gate), `is_running`/`is_healthy`/`stop`,
`url()` (host-facing) vs `gateway_url()` (what the gateway resolves against),
and a concrete `mint_gateway_token()` (the orchestrator holds the signing key,
so it issues the gateway's token — #469).

`DockerOrchestrator` (backend/docker/orchestrator.py) is the first impl,
extracted out of `DockerInfraService`: the control-plane container run, the
`--internal` control network, the source-hash recreate gate, health polling,
and the orchestrator constants (name/label/network/image/dockerfile/hash-label).
`DockerInfraService` now composes `orchestrator()` + `gateway()` — each builds
its own image via `ensure_built`, the orchestrator comes up first, then the
gateway is connected with the orchestrator-minted token. Its `url`/`is_healthy`
delegate to the orchestrator service.

Split the container-lifecycle tests into test_docker_orchestrator; test_docker_infra
now covers the composition. macOS + firecracker follow.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-25 14:36:35 -04:00
parent 57d32d2c5c
commit cc4e29c3da
6 changed files with 581 additions and 354 deletions
+35 -180
View File
@@ -1,209 +1,64 @@
"""Unit: orchestrator + gateway container lifecycle (PRD 0070 plane split).
"""Unit: DockerInfraService composes the orchestrator + gateway services.
`DockerInfraService` brings up two containers — the lean orchestrator on the
`--internal` control network + host loopback, and the dual-homed gateway. These
tests isolate the orchestrator-container logic by mocking `gateway` (the
gateway runs via `DockerGateway`, exercised in its own test)."""
`DockerInfraService` no longer owns the container lifecycles — it composes the
`DockerOrchestrator` (control plane) and `DockerGateway` (data plane) services,
each tested in its own module. These tests isolate the *composition*: both are
built, the orchestrator comes up first, then the gateway is connected to it with
a freshly minted token."""
from __future__ import annotations
import tempfile
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import MagicMock, patch
from bot_bottle.gateway import GatewayError
from bot_bottle.backend.docker.infra import (
ORCHESTRATOR_NAME,
ORCHESTRATOR_NETWORK,
ORCHESTRATOR_SOURCE_HASH_LABEL,
DockerInfraService,
)
from bot_bottle.gateway import GATEWAY_NAME
from bot_bottle.orchestrator.lifecycle import (
OrchestratorStartError,
source_hash,
)
from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.backend.docker.infra.urllib.request.urlopen"
_RUN = "bot_bottle.backend.docker.infra.run_docker"
_SLEEP = "bot_bottle.backend.docker.infra.time.sleep"
_MONOTONIC = "bot_bottle.backend.docker.infra.time.monotonic"
def _health(status: int) -> MagicMock:
m = MagicMock()
m.__enter__.return_value.status = status
return m
def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
class TestDockerInfraService(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.svc = DockerInfraService(port=8099)
# Isolate the orchestrator-container logic: the gateway is brought up by
# DockerGateway (its own module/run_docker), tested separately.
self.orch = MagicMock()
self.orch.url.return_value = "http://127.0.0.1:8099"
self.orch.gateway_url.return_value = f"http://{ORCHESTRATOR_NAME}:8099"
self.orch.mint_gateway_token.return_value = "gw.jwt"
self.gw = MagicMock()
patcher = patch.object(self.svc, "gateway", return_value=self.gw)
patcher.start()
self.addCleanup(patcher.stop)
def test_url(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.svc.url)
for name, mock in (("orchestrator", self.orch), ("gateway", self.gw)):
p = patch.object(self.svc, name, return_value=mock)
p.start()
self.addCleanup(p.stop)
def test_gateway_name(self) -> None:
self.assertEqual(GATEWAY_NAME, self.svc.gateway_name)
def test_is_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.svc.is_healthy())
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.svc.is_healthy())
def test_url_delegates_to_the_orchestrator(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.svc.url)
def test_noop_when_healthy_and_source_unchanged(self) -> None:
# A healthy orchestrator on current source is left alone — recreating it
# on every launch drops in-memory egress tokens (#381). The gateway is
# still refreshed (idempotent).
current = source_hash(self.svc._repo_root)
def test_is_healthy_delegates_to_the_orchestrator(self) -> None:
self.orch.is_healthy.return_value = True
self.assertTrue(self.svc.is_healthy())
self.orch.is_healthy.assert_called_once()
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout=current)
return _proc()
def test_ensure_running_builds_both_then_starts_orchestrator_first(self) -> None:
url = self.svc.ensure_running()
self.assertEqual("http://127.0.0.1:8099", url)
# Both images built; the control plane comes up before the gateway.
self.orch.ensure_built.assert_called_once()
self.gw.ensure_built.assert_called_once()
self.orch.ensure_running.assert_called_once()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual([], runs)
self.gw.connect_to_orchestrator.assert_called_once()
def test_recreates_orchestrator_when_source_changed(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout="stale-hash")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
self.assertIn(ORCHESTRATOR_NAME, runs[0])
current = source_hash(self.svc._repo_root)
self.assertIn(f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current}", runs[0])
def test_starts_orchestrator_when_absent(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
argv = runs[0]
self.assertIn(ORCHESTRATOR_NAME, argv)
# Control network only — agents are never on it.
self.assertEqual(ORCHESTRATOR_NETWORK, argv[argv.index("--network") + 1])
# Published on loopback for the CLI, mapped to the fixed internal 8099.
self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
# The lean control plane: no mitmproxy CA mount, no gateway daemons.
self.assertFalse([a for a in argv if a.endswith(":/home/mitmproxy/.mitmproxy")])
self.assertNotIn("BOT_BOTTLE_GATEWAY_DAEMONS", " ".join(argv))
# Orchestrator entrypoint args (image ENTRYPOINT is `-m bot_bottle.orchestrator`).
self.assertIn("--broker", argv)
self.assertIn("stub", argv)
# Gateway is brought up after the control plane is healthy.
self.gw.connect_to_orchestrator.assert_called_once()
def test_builds_gateway_and_orchestrator_images(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.svc.ensure_running()
builds = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "build"]]
# Two images now (the combined Dockerfile.infra is retired).
self.assertEqual(2, len(builds))
dockerfiles = [next(a for a in b if "Dockerfile" in a) for b in builds]
self.assertIn("Dockerfile.gateway", dockerfiles[0])
self.assertIn("Dockerfile.orchestrator", dockerfiles[1])
def test_publish_maps_host_port_to_fixed_internal_port(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
svc = DockerInfraService(port=20001)
with patch.object(svc, "gateway", return_value=MagicMock()), \
patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
svc.ensure_running()
argv = next(c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"])
self.assertEqual("127.0.0.1:20001:8099", argv[argv.index("--publish") + 1])
def test_raises_on_control_plane_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_RUN, return_value=_proc()), \
patch(_SLEEP), patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.svc.ensure_running(startup_timeout=1.0)
def test_noop_when_healthy_and_inspect_fails(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(returncode=1, stderr="daemon error")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.svc.ensure_running()
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual([], runs)
def test_build_failure_raises(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_RUN, return_value=_proc(returncode=1, stderr="no space left")):
with self.assertRaises(GatewayError):
self.svc.ensure_running()
def test_creates_control_network_internal(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="not found")
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.svc.ensure_running()
creates = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "network", "create"]]
# The control network is created --internal; the data network isn't.
control = [c for c in creates if ORCHESTRATOR_NETWORK in c]
self.assertEqual(1, len(control))
self.assertIn("--internal", control[0])
def test_ensure_running_connects_gateway_with_minted_token(self) -> None:
self.svc.ensure_running()
# The gateway is bound to the orchestrator's *gateway_url* (the control
# DNS name), carrying the orchestrator-minted `gateway` token.
self.gw.connect_to_orchestrator.assert_called_once_with(
f"http://{ORCHESTRATOR_NAME}:8099", "gw.jwt")
self.orch.mint_gateway_token.assert_called_once()
def test_stop_removes_both_containers(self) -> None:
with patch(_RUN) as run:
+196
View File
@@ -0,0 +1,196 @@
"""Unit: the docker orchestrator (control plane) container lifecycle (PRD 0070)."""
from __future__ import annotations
import unittest
import urllib.error
from unittest.mock import MagicMock, Mock, patch
from bot_bottle.backend.docker.orchestrator import (
ORCHESTRATOR_NAME,
ORCHESTRATOR_NETWORK,
ORCHESTRATOR_SOURCE_HASH_LABEL,
DockerOrchestrator,
)
from bot_bottle.gateway import GatewayError
from bot_bottle.orchestrator.lifecycle import OrchestratorStartError, source_hash
_ORCH = "bot_bottle.backend.docker.orchestrator"
_RUN = f"{_ORCH}.run_docker"
_SLEEP = f"{_ORCH}.time.sleep"
_MONOTONIC = f"{_ORCH}.time.monotonic"
_TOKEN = f"{_ORCH}.host_orchestrator_token"
# The ABC's is_healthy probes /health via urllib in the lifecycle module.
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
def _health(status: int) -> MagicMock:
m = MagicMock()
m.__enter__.return_value.status = status
return m
def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
class TestDockerOrchestrator(unittest.TestCase):
def setUp(self) -> None:
self.orch = DockerOrchestrator(port=8099)
# _run_container seeds the signing key; keep it off the real host file.
p = patch(_TOKEN, return_value="signing-key")
p.start()
self.addCleanup(p.stop)
def test_url_is_host_loopback(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.orch.url())
def test_gateway_url_is_the_container_dns_name(self) -> None:
# The gateway reaches the orchestrator by name on the control network.
self.assertEqual(f"http://{ORCHESTRATOR_NAME}:8099", self.orch.gateway_url())
def test_is_healthy_probes_health_on_the_loopback_url(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.orch.is_healthy())
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.orch.is_healthy())
def test_noop_when_healthy_and_source_unchanged(self) -> None:
# A healthy orchestrator on current source is left alone — recreating it
# on every launch drops in-memory egress tokens (#381).
current = source_hash(self.orch._repo_root)
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout=current)
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.orch.ensure_running()
self.assertEqual([], [c.args[0] for c in run.call_args_list
if c.args[0][:2] == ["docker", "run"]])
def test_recreates_when_source_changed(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout="stale-hash")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.orch.ensure_running()
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
self.assertIn(ORCHESTRATOR_NAME, runs[0])
current = source_hash(self.orch._repo_root)
self.assertIn(f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current}", runs[0])
def test_starts_when_absent_on_control_net_and_loopback(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.orch.ensure_running()
argv = next(c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"])
self.assertIn(ORCHESTRATOR_NAME, argv)
# Control network only — agents are never on it.
self.assertEqual(ORCHESTRATOR_NETWORK, argv[argv.index("--network") + 1])
# Published on loopback for the CLI, mapped to the fixed internal 8099.
self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
# The lean control plane: no mitmproxy CA mount, no gateway daemons.
self.assertFalse([a for a in argv if a.endswith(":/home/mitmproxy/.mitmproxy")])
self.assertNotIn("BOT_BOTTLE_GATEWAY_DAEMONS", " ".join(argv))
# Orchestrator entrypoint args (image ENTRYPOINT is `-m bot_bottle.orchestrator`).
self.assertIn("--broker", argv)
self.assertIn("stub", argv)
def test_publish_maps_host_port_to_fixed_internal_port(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
orch = DockerOrchestrator(port=20001)
with patch(_TOKEN, return_value="k"), \
patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
orch.ensure_running()
argv = next(c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"])
self.assertEqual("127.0.0.1:20001:8099", argv[argv.index("--publish") + 1])
def test_raises_on_startup_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_RUN, return_value=_proc()), \
patch(_SLEEP), patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.orch.ensure_running(startup_timeout=1.0)
def test_noop_when_healthy_and_inspect_fails(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(returncode=1, stderr="daemon error")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.orch.ensure_running()
self.assertEqual([], [c.args[0] for c in run.call_args_list
if c.args[0][:2] == ["docker", "run"]])
def test_creates_control_network_internal(self) -> None:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="not found")
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake) as run, patch(_SLEEP):
self.orch.ensure_running()
creates = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "network", "create"]]
control = [c for c in creates if ORCHESTRATOR_NETWORK in c]
self.assertEqual(1, len(control))
self.assertIn("--internal", control[0])
def test_ensure_built_builds_the_orchestrator_image(self) -> None:
with patch(_RUN, return_value=_proc()) as run:
self.orch.ensure_built()
builds = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "build"]]
self.assertEqual(1, len(builds))
self.assertTrue(any(a.endswith("Dockerfile.orchestrator") for a in builds[0]))
def test_ensure_built_raises_on_build_failure(self) -> None:
with patch(_RUN, return_value=_proc(returncode=1, stderr="no space left")):
with self.assertRaises(GatewayError):
self.orch.ensure_built()
def test_ensure_built_noop_without_dockerfile(self) -> None:
orch = DockerOrchestrator(dockerfile=None)
with patch(_RUN) as run:
orch.ensure_built()
run.assert_not_called()
def test_is_running_reads_docker_ps(self) -> None:
with patch(_RUN, return_value=_proc(stdout=ORCHESTRATOR_NAME)):
self.assertTrue(self.orch.is_running())
with patch(_RUN, return_value=_proc(stdout="")):
self.assertFalse(self.orch.is_running())
def test_stop_removes_the_container(self) -> None:
with patch(_RUN) as run:
self.orch.stop()
argv = run.call_args.args[0]
self.assertEqual(["docker", "rm", "--force", ORCHESTRATOR_NAME], argv)
if __name__ == "__main__":
unittest.main()