Merge pull request #23 from openqj/tenant-isolation
Multi-tenant isolation for Docker tasks
This commit is contained in:
5 files changed
+457
-26
No files matched your search
@@ -479,6 +479,18 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
|
||||
- 清理与取消:`docker run --rm`,正常失败也自动删容器;超时
|
||||
时先 docker kill 再 docker rm -f;worker 取消时按任务派生的
|
||||
容器名 docker kill,不只改数据库状态。
|
||||
- 多租户隔离(`--tenant-isolation`,仅 Docker 执行器):开启
|
||||
后每个项目用独立的宿主目录 `tenants_root/<项目>/work`、
|
||||
`codex-home`、test-repo 副本与 repo 检出,容器内路径与
|
||||
非租户模式完全一致(`/app/work`、codex-home、repos/<项目>、
|
||||
test-repo),只是宿主挂载源按项目分开——容器内看不到其它
|
||||
项目的任何文件;同时每个项目进独立的 Docker 网络
|
||||
`codex-<项目>`(容器跨项目网络隔离),容器统一加
|
||||
`--cap-drop=ALL --security-opt=no-new-privileges`。codex-home
|
||||
只播种模板的 config.toml(relay 端点与 provider 名,不含
|
||||
密钥值),绝不复制 thread_history。注意:内层 codex 沙箱仍
|
||||
是 danger-full-access(主机内核禁 user namespace,bwrap 不可
|
||||
用),隔离边界是容器 + 挂载 + 网络三层,不是内层沙箱。
|
||||
- 网络白名单的诚实边界:docker run 默认 bridge 网络,本执行
|
||||
器不做 egress 白名单——真正的“只放行模型 API”需要宿主防火
|
||||
墙或代理规则,那是部署项,不是本轮代码假装做到的,不要宣称
|
||||
|
||||
+67
-14
@@ -369,13 +369,24 @@ class DockerExecutor(Executor):
|
||||
"Missing environment variable: %s" % self.env_key)
|
||||
|
||||
# -- 命令构造 ------------------------------------------------------
|
||||
def build_run_command(self, task, inner_cmd):
|
||||
def build_run_command(self, task, inner_cmd, extra_mounts=None,
|
||||
network=None):
|
||||
"""构造完整的 docker run 命令(纯函数,可单测断言)。
|
||||
|
||||
顺序固定:docker run --rm --name <容器名> --user <用户>
|
||||
--cpus <核数> --memory <内存> -v <worktree>:/work
|
||||
-v <代码目录>:/app:ro -w /work [-e <env_key名>]
|
||||
<镜像> <容器内命令...>。-e 只传名字,不带 =值。
|
||||
--cpus <核数> --memory <内存> [--cap-drop=ALL
|
||||
--security-opt=no-new-privileges] [--network <网络>]
|
||||
-v <worktree>:/work -v <代码目录>:/app:ro -w /work
|
||||
[-e <env_key名>] <镜像> <容器内命令...>。-e 只传名字,
|
||||
不带 =值。
|
||||
extra_mounts 给定时替换构造时的静态挂载(多租户隔离
|
||||
用:worker 按项目算出挂载后每次调用传入);不给时用
|
||||
静态挂载。network 给定时容器进该 Docker 网络(多租户
|
||||
用项目独立网络,容器之间默认 bridge 上互相可见的问题
|
||||
由此解决);不给时用 docker 默认网络。
|
||||
降权:容器默认已是非 root 用户,再加 --cap-drop=ALL
|
||||
与 no-new-privileges(提权类工具调用会被挡下,codex
|
||||
的常规文件/git 操作不受影响,见多租户实测)。
|
||||
"""
|
||||
self._check_docker()
|
||||
self._check_key()
|
||||
@@ -384,13 +395,19 @@ class DockerExecutor(Executor):
|
||||
"--user", self.user,
|
||||
"--cpus", self.cpus,
|
||||
"--memory", self.memory,
|
||||
"-v", "%s:/work" % self._worktree_for(task),
|
||||
"-v", "%s:/app:ro" % self._app_dir(),
|
||||
"-w", "/work"]
|
||||
"--cap-drop=ALL",
|
||||
"--security-opt=no-new-privileges"]
|
||||
if network:
|
||||
cmd += ["--network", network]
|
||||
cmd += ["-v", "%s:/work" % self._worktree_for(task),
|
||||
"-v", "%s:/app:ro" % self._app_dir(),
|
||||
"-w", "/work"]
|
||||
# 额外挂载放在基础挂载之后:/app/work 这类子路径挂载会
|
||||
# 覆盖 /app:ro 的对应子路径,让容器内 host.py 能写
|
||||
# work/ 和 host-state.json,同时代码主体仍只读。
|
||||
for host_p, cont_p, mode in self.extra_mounts:
|
||||
mounts = self.extra_mounts if extra_mounts is None \
|
||||
else self._parse_extra_mounts(extra_mounts)
|
||||
for host_p, cont_p, mode in mounts:
|
||||
cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)]
|
||||
if self.env_key:
|
||||
cmd += ["-e", self.env_key]
|
||||
@@ -500,18 +517,22 @@ class DockerExecutor(Executor):
|
||||
continue
|
||||
return ok
|
||||
|
||||
def _run_in_container(self, task, inner_cmd, timeout=None):
|
||||
def _run_in_container(self, task, inner_cmd, timeout=None,
|
||||
extra_mounts=None, network=None):
|
||||
"""在容器里跑一轮内层命令,返回 LocalExecutor 同款结果 dict。
|
||||
|
||||
超时:runner 抛 subprocess.TimeoutExpired(或同名异常)
|
||||
时先 kill+rm 容器,再返回 failed(stdout 截断 + timeout
|
||||
说明,returncode 为 None);其它 runner 异常也如实返回
|
||||
failed,不编造成功。timeout 只影响本次调用,不改实例。
|
||||
extra_mounts/network 只影响本次调用(多租户按项目传入)。
|
||||
"""
|
||||
task_id = task.get("id") if isinstance(task, dict) else None
|
||||
if task_id is None:
|
||||
raise ValueError("任务缺少 id")
|
||||
cmd = self.build_run_command(task, inner_cmd)
|
||||
cmd = self.build_run_command(task, inner_cmd,
|
||||
extra_mounts=extra_mounts,
|
||||
network=network)
|
||||
try:
|
||||
result = self._call_runner(cmd, timeout=timeout)
|
||||
except Exception as exc:
|
||||
@@ -536,7 +557,8 @@ class DockerExecutor(Executor):
|
||||
parsed["container"] = self.container_name(task)
|
||||
return parsed
|
||||
|
||||
def dispatch_task(self, task, timeout=None) -> dict:
|
||||
def dispatch_task(self, task, timeout=None, extra_mounts=None,
|
||||
network=None) -> dict:
|
||||
"""容器内调 host.py dispatch:建 host 任务并跑一轮 turn。"""
|
||||
if not isinstance(task, dict) or task.get("id") is None:
|
||||
raise ValueError("任务缺少 id")
|
||||
@@ -548,9 +570,12 @@ class DockerExecutor(Executor):
|
||||
repo = task.get("repo")
|
||||
if repo:
|
||||
inner += ["--repo", repo]
|
||||
return self._run_in_container(task, inner, timeout=timeout)
|
||||
return self._run_in_container(task, inner, timeout=timeout,
|
||||
extra_mounts=extra_mounts,
|
||||
network=network)
|
||||
|
||||
def run_task(self, task, timeout=None) -> dict:
|
||||
def run_task(self, task, timeout=None, extra_mounts=None,
|
||||
network=None) -> dict:
|
||||
"""容器内调 host.py run:在已有 host 任务上再跑一轮 turn。"""
|
||||
if not isinstance(task, dict) or task.get("id") is None:
|
||||
raise ValueError("任务缺少 id")
|
||||
@@ -559,7 +584,35 @@ class DockerExecutor(Executor):
|
||||
inner = self._inner_host_command(
|
||||
"run", "--task", self._host_name(task),
|
||||
"--prompt", task["prompt"], model=task.get("model"))
|
||||
return self._run_in_container(task, inner, timeout=timeout)
|
||||
return self._run_in_container(task, inner, timeout=timeout,
|
||||
extra_mounts=extra_mounts,
|
||||
network=network)
|
||||
|
||||
def ensure_network(self, name):
|
||||
"""确保 Docker 网络存在:不存在就创建,已存在就忽略。
|
||||
|
||||
返回 True 表示网络可用(创建成功或本来就存在),
|
||||
False 表示创建失败(调用方继续派发,docker run 时会
|
||||
因网络不存在而明确失败,不在这里静默降级默认网络——
|
||||
降级会让隔离悄悄失效,比失败更糟,所以这里只返回
|
||||
布尔值,是否继续由调用方按失败处理:worker 在租户
|
||||
模式下网络创建失败时直接让任务 failed,不跑容器)。
|
||||
"""
|
||||
if not name:
|
||||
return False
|
||||
try:
|
||||
result = self._call_runner(["docker", "network", "create", name],
|
||||
timeout=30)
|
||||
except Exception as exc:
|
||||
return "already exists" in str(exc)
|
||||
try:
|
||||
if getattr(result, "returncode", 1) == 0:
|
||||
return True
|
||||
text = (getattr(result, "stderr", "") or "") + \
|
||||
(getattr(result, "stdout", "") or "")
|
||||
return "already exists" in text
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def execute(self, task) -> dict:
|
||||
"""只读执行:容器内调 host.py status,不直接跑模型回合。
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
"""多租户隔离单测。
|
||||
|
||||
只用临时目录与假 runner,不跑真实 Docker、不联网。
|
||||
覆盖:租户目录播种(codex-home 只取 config.toml,绝不
|
||||
复制 thread_history)、挂载按项目分开、网络名安全化、
|
||||
执行器 per-call 挂载/网络覆盖、网络创建失败不降级。
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
try:
|
||||
from cloud.sandbox import DockerExecutor
|
||||
from cloud.server import init_db
|
||||
from cloud.worker import Worker
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.sandbox import DockerExecutor
|
||||
from cloud.server import init_db
|
||||
from cloud.worker import Worker
|
||||
|
||||
|
||||
def _ok_runner_factory(seen):
|
||||
def fake(cmd, timeout=None):
|
||||
seen.append(list(cmd))
|
||||
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||
return fake
|
||||
|
||||
|
||||
class TenantDirsTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
init_db(self.db_path)
|
||||
# 模板 home:config.toml + 绝不能复制的 thread_history
|
||||
# 与假密钥文件;模板 repo:sample.py + .git。
|
||||
self.template_home = os.path.join(self.tmp.name, "tpl-home")
|
||||
os.makedirs(self.template_home)
|
||||
open(os.path.join(self.template_home, "config.toml"),
|
||||
"w").write('[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n')
|
||||
open(os.path.join(self.template_home, "thread_history_1.sqlite"),
|
||||
"w").write("OTHER TENANT HISTORY")
|
||||
open(os.path.join(self.template_home, "auth.json"),
|
||||
"w").write("SECRET")
|
||||
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
|
||||
os.makedirs(os.path.join(self.template_repo, ".git"))
|
||||
open(os.path.join(self.template_repo, "sample.py"),
|
||||
"w").write("def add(a, b):\n return a + b\n")
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _worker(self, **kwargs):
|
||||
kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
||||
kwargs.setdefault("tenants_root",
|
||||
os.path.join(self.tmp.name, "tenants"))
|
||||
kwargs.setdefault("tenant_template_home", self.template_home)
|
||||
kwargs.setdefault("tenant_template_repo", self.template_repo)
|
||||
kwargs.setdefault("tenant_isolation", True)
|
||||
return Worker(self.db_path, "/fake/host.py", **kwargs)
|
||||
|
||||
def test_dirs_seed_config_only_not_history(self):
|
||||
worker = self._worker()
|
||||
dirs = worker._tenant_dirs("projA")
|
||||
home = dirs["codex_home"]
|
||||
self.assertTrue(os.path.exists(os.path.join(home, "config.toml")))
|
||||
self.assertFalse(os.path.exists(
|
||||
os.path.join(home, "thread_history_1.sqlite")))
|
||||
self.assertFalse(os.path.exists(os.path.join(home, "auth.json")))
|
||||
# test-repo 副本有 sample.py,且是复制不是链接。
|
||||
self.assertTrue(os.path.exists(
|
||||
os.path.join(dirs["test_repo"], "sample.py")))
|
||||
# 第二次调用不覆盖租户自己的改动。
|
||||
marker = os.path.join(dirs["test_repo"], "mine.txt")
|
||||
open(marker, "w").write("mine")
|
||||
worker._tenant_dirs("projA")
|
||||
self.assertTrue(os.path.exists(marker))
|
||||
|
||||
def test_mounts_are_per_project(self):
|
||||
worker = self._worker()
|
||||
mounts_a = worker._tenant_mounts("projA")
|
||||
mounts_b = worker._tenant_mounts("projB")
|
||||
# 容器内路径相同(/app/work 等),宿主源不同项目不同。
|
||||
cont_a = [m.split(":")[1] for m in mounts_a]
|
||||
cont_b = [m.split(":")[1] for m in mounts_b]
|
||||
self.assertEqual(cont_a, cont_b)
|
||||
self.assertIn("/app/work", cont_a)
|
||||
self.assertIn("/var/lib/codex-app-server/codex-home", cont_a)
|
||||
hosts_a = {m.split(":")[0] for m in mounts_a}
|
||||
hosts_b = {m.split(":")[0] for m in mounts_b}
|
||||
self.assertFalse(hosts_a & hosts_b)
|
||||
# 租户 A 的 work 里写文件,B 的挂载里看不到(宿主侧)。
|
||||
open(os.path.join(worker._tenant_dirs("projA")["work"],
|
||||
"secret-a.txt"), "w").write("SECRET-A")
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
worker._tenant_dirs("projB")["work"], "secret-a.txt")))
|
||||
|
||||
def test_network_name_sanitized(self):
|
||||
worker = self._worker()
|
||||
self.assertEqual(worker._tenant_network("abc123"),
|
||||
"codex-abc123")
|
||||
self.assertEqual(worker._tenant_network("a b/c"),
|
||||
"codex-a-b-c")
|
||||
with self.assertRaises(ValueError):
|
||||
worker._tenant_network("...")
|
||||
|
||||
def test_template_home_without_config_fails_loudly(self):
|
||||
empty_home = os.path.join(self.tmp.name, "empty-home")
|
||||
os.makedirs(empty_home)
|
||||
worker = self._worker(tenant_template_home=empty_home)
|
||||
with self.assertRaises(RuntimeError):
|
||||
worker._tenant_dirs("projA")
|
||||
|
||||
|
||||
class DockerTenantCommandTestCase(unittest.TestCase):
|
||||
def test_per_call_mounts_network_and_drop_caps(self):
|
||||
seen = []
|
||||
static_dir = tempfile.mkdtemp()
|
||||
tenant_dir = tempfile.mkdtemp()
|
||||
with mock.patch("cloud.sandbox.shutil.which",
|
||||
return_value="/usr/bin/docker"):
|
||||
ex = DockerExecutor(
|
||||
"/fake/host.py",
|
||||
runner=_ok_runner_factory(seen),
|
||||
env={"XCM_API_KEY": "test-placeholder"},
|
||||
extra_mounts=["%s:/app/work:rw" % static_dir])
|
||||
cmd = ex.build_run_command(
|
||||
{"id": "t1", "worktree": "/tmp/wt"},
|
||||
["echo", "hi"],
|
||||
extra_mounts=["%s:/app/work:rw" % tenant_dir],
|
||||
network="codex-projA")
|
||||
mounts = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-v"]
|
||||
# per-call 挂载替换静态挂载,不是追加。
|
||||
self.assertIn("%s:/app/work:rw" % tenant_dir, mounts)
|
||||
self.assertNotIn("%s:/app/work:rw" % static_dir, mounts)
|
||||
self.assertEqual(cmd[cmd.index("--network") + 1], "codex-projA")
|
||||
self.assertIn("--cap-drop=ALL", cmd)
|
||||
self.assertIn("--security-opt=no-new-privileges", cmd)
|
||||
|
||||
def test_ensure_network_already_exists_is_ok(self):
|
||||
def exists_runner(cmd, timeout=None):
|
||||
return SimpleNamespace(
|
||||
returncode=1, stdout="",
|
||||
stderr="Error response from daemon: network with name "
|
||||
"codex-p already exists")
|
||||
with mock.patch("cloud.sandbox.shutil.which",
|
||||
return_value="/usr/bin/docker"):
|
||||
ex = DockerExecutor("/fake/host.py", runner=exists_runner,
|
||||
env={"XCM_API_KEY": "test-placeholder"})
|
||||
self.assertTrue(ex.ensure_network("codex-p"))
|
||||
|
||||
def test_ensure_network_real_failure_is_false(self):
|
||||
def fail_runner(cmd, timeout=None):
|
||||
return SimpleNamespace(returncode=1, stdout="",
|
||||
stderr="daemon not reachable")
|
||||
with mock.patch("cloud.sandbox.shutil.which",
|
||||
return_value="/usr/bin/docker"):
|
||||
ex = DockerExecutor("/fake/host.py", runner=fail_runner,
|
||||
env={"XCM_API_KEY": "test-placeholder"})
|
||||
self.assertFalse(ex.ensure_network("codex-p"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+208
-11
@@ -233,7 +233,9 @@ class Worker:
|
||||
task_timeout=1800, max_retries=1,
|
||||
retry_backoff_seconds=5.0, repos_dir="./repos",
|
||||
repos_runner=None, heartbeat_file=None,
|
||||
worker_id=None, capacity=1, fallback_models=None):
|
||||
worker_id=None, capacity=1, fallback_models=None,
|
||||
tenant_isolation=False, tenants_root=None,
|
||||
tenant_template_home=None, tenant_template_repo=None):
|
||||
self.db_path = str(db_path)
|
||||
self.host_py_path = host_py_path
|
||||
self.config_path = config_path
|
||||
@@ -298,6 +300,22 @@ class Worker:
|
||||
# repos_runner 可注入(单测用假 git runner),缺省走真实 git。
|
||||
self.repos_dir = repos_dir
|
||||
self.repos_runner = repos_runner
|
||||
# 多租户隔离(仅 Docker 执行器有效,Local 执行器在本机
|
||||
# 直接跑 host.py,同一用户下本来就没有容器边界,flag
|
||||
# 对它不起作用):开启后每个项目用独立的宿主目录
|
||||
# tenants_root/<project_id>/{work,codex-home,test-repo,
|
||||
# <project_id>},容器内路径保持与原来完全一致(/app/work、
|
||||
# /var/lib/.../codex-home 等),只是宿主侧的挂载源按
|
||||
# 项目分开——容器内看不到其它项目的任何文件。codex-home
|
||||
# 只播种 config.toml(relay 配置,不含任何密钥值),
|
||||
# 绝不复制 thread_history(那是其它任务的 prompt 与结果)。
|
||||
self.tenant_isolation = tenant_isolation
|
||||
self.tenants_root = tenants_root or \
|
||||
"/var/lib/codex-app-server/tenants"
|
||||
self.tenant_template_home = tenant_template_home or \
|
||||
"/var/lib/codex-app-server/codex-home"
|
||||
self.tenant_template_repo = tenant_template_repo or \
|
||||
"/var/lib/codex-app-server/test-repo"
|
||||
# 运维心跳:路径优先显式参数,其次环境变量,默认当前目录下
|
||||
# 的 worker-heartbeat.json;主机路径不在代码里硬编码,主机
|
||||
# 部署时用 --heartbeat-file 或环境变量传入(见 deploy/README)。
|
||||
@@ -856,6 +874,107 @@ class Worker:
|
||||
"stdout": out or "", "stderr": err or "",
|
||||
"returncode": proc.returncode, "host_status": None}
|
||||
|
||||
# -- 多租户隔离 ----------------------------------------------------
|
||||
def _tenant_safe_name(self, project_id):
|
||||
"""项目 id 的安全目录名:与 repos.checkout_dir 同规则。"""
|
||||
if project_id is None:
|
||||
raise ValueError("project_id 缺失,无法确定租户目录")
|
||||
raw = str(project_id)
|
||||
safe = re.sub(r"[^A-Za-z0-9_.-]", "-", raw).strip(".")
|
||||
if not safe or safe in (".", ".."):
|
||||
raise ValueError("project_id 无法映射为安全的租户目录名: %r"
|
||||
% raw)
|
||||
return safe
|
||||
|
||||
def _tenant_dirs(self, project_id):
|
||||
"""某项目的租户宿主目录(不存在时创建并播种)。
|
||||
|
||||
返回 dict:work / codex_home / repo_checkout / test_repo
|
||||
(均为宿主绝对路径)。播种规则:
|
||||
- codex-home:只复制模板的 config.toml(relay 端点与
|
||||
provider 名,不含密钥值);模板缺 config.toml 时直接
|
||||
报错——空 home 会让 codex 去 api.openai.com 且无 auth,
|
||||
绝不能静默放行(见 PR #18 的教训)。
|
||||
- test-repo:整目录复制模板(它是 sample.py 夹具,含
|
||||
.git 历史也无妨,本来就是公开的起始代码)。
|
||||
- work / repo_checkout:只建目录。
|
||||
并发下两个同项目任务同时播种是安全的:mkdir 用
|
||||
exist_ok,复制前判存在且复制内容相同(模板),即使
|
||||
race 到重复复制也不会写坏租户自己的数据(租户数据
|
||||
在复制完成后才产生)。
|
||||
"""
|
||||
import shutil
|
||||
|
||||
safe = self._tenant_safe_name(project_id)
|
||||
base = os.path.join(self.tenants_root, safe)
|
||||
dirs = {
|
||||
"work": os.path.join(base, "work"),
|
||||
"codex_home": os.path.join(base, "codex-home"),
|
||||
"repo_checkout": os.path.join(base, safe),
|
||||
"test_repo": os.path.join(base, "test-repo"),
|
||||
}
|
||||
# 注意绝不 mkdir repo_checkout:repo_url 任务的检出由
|
||||
# resolve_repo 负责 clone,空目录先占位会导致它误判
|
||||
# “目录已存在但不是 git 仓库”而拒绝 clone;_tenant_mounts
|
||||
# 里按该目录是否存在决定是否挂载它(不存在说明任务
|
||||
# 不用它,走 test-repo)。
|
||||
for key in ("work", "codex_home", "test_repo"):
|
||||
os.makedirs(dirs[key], exist_ok=True)
|
||||
home_config = os.path.join(dirs["codex_home"], "config.toml")
|
||||
if not os.path.exists(home_config):
|
||||
template = os.path.join(self.tenant_template_home,
|
||||
"config.toml")
|
||||
if not os.path.exists(template):
|
||||
raise RuntimeError(
|
||||
"租户 codex-home 模板缺 config.toml:%s(空 home"
|
||||
" 会让 codex 去 api.openai.com 且无 auth,拒绝"
|
||||
" 派发)" % template)
|
||||
shutil.copyfile(template, home_config)
|
||||
if not os.path.exists(os.path.join(dirs["test_repo"], ".git")) \
|
||||
and not os.path.exists(dirs["test_repo"] + "/sample.py"):
|
||||
# 目标为空目录(刚 mkdir 出来的)才播种;已有内容
|
||||
# 时绝不覆盖(那是该项目自己的改动或上次播种结果)。
|
||||
if not os.listdir(dirs["test_repo"]):
|
||||
if not os.path.exists(self.tenant_template_repo):
|
||||
raise RuntimeError(
|
||||
"租户 test-repo 模板不存在:%s"
|
||||
% self.tenant_template_repo)
|
||||
os.rmdir(dirs["test_repo"])
|
||||
shutil.copytree(self.tenant_template_repo,
|
||||
dirs["test_repo"])
|
||||
return dirs
|
||||
|
||||
def _tenant_mounts(self, project_id):
|
||||
"""某项目任务的 Docker 挂载列表(字符串,供执行器解析)。
|
||||
|
||||
容器内路径与非租户模式完全一致,只是宿主源按项目分开:
|
||||
/app/work、codex-home、repos/<pid>、test-repo 全部指向
|
||||
该项目的租户目录;代码目录 /app:ro 由执行器基础挂载
|
||||
提供,不在这里重复。静态 --docker-extra-mounts 在租户
|
||||
模式下被这份列表替换(否则共享挂载会继续泄漏)。
|
||||
repos 挂载只在检出目录已存在时才加(repo_url 任务在
|
||||
派发前已由 _apply_project_repo clone 好;test-repo 任务
|
||||
不用它,不存在时硬挂载会被执行器的存在性检查拒绝)。
|
||||
"""
|
||||
dirs = self._tenant_dirs(project_id)
|
||||
mounts = [
|
||||
"%s:/app/work:rw" % dirs["work"],
|
||||
"%s:/var/lib/codex-app-server/codex-home:rw"
|
||||
% dirs["codex_home"],
|
||||
"%s:/var/lib/codex-app-server/test-repo:rw"
|
||||
% dirs["test_repo"],
|
||||
]
|
||||
if os.path.exists(dirs["repo_checkout"]):
|
||||
mounts.append(
|
||||
"%s:/var/lib/codex-app-server/repos/%s:rw"
|
||||
% (dirs["repo_checkout"],
|
||||
self._tenant_safe_name(project_id)))
|
||||
return mounts
|
||||
|
||||
def _tenant_network(self, project_id):
|
||||
"""某项目的 Docker 网络名(容器跨项目网络隔离用)。"""
|
||||
return "codex-" + self._tenant_safe_name(project_id)
|
||||
|
||||
# -- 项目仓库 ------------------------------------------------------
|
||||
def _project_repo_url(self, project_id):
|
||||
"""查任务所属项目的 repo_url;查不到项目、列缺失、表缺失时
|
||||
@@ -882,21 +1001,43 @@ class Worker:
|
||||
def _apply_project_repo(self, task):
|
||||
"""派发前按项目 repo_url 解析仓库,返回错误文本或 None。
|
||||
|
||||
repo_url 为空:不动 task,返回 None(走原有默认 repo)。
|
||||
repo_url 为空:非租户模式不动 task,返回 None(走原有
|
||||
默认 repo);租户模式则把 task["repo"] 指到该项目的
|
||||
test-repo 副本(容器内路径与原来一致,宿主源已按项目
|
||||
分开),否则容器会 fallback 到共享 test-repo。
|
||||
repo_url 非空:调 cloud.repos.resolve_repo 得到本地检出,
|
||||
设 task["repo"] 再交给执行器——LocalExecutor 与
|
||||
DockerExecutor 的 dispatch_task 都已读 task["repo"],同一条
|
||||
路径,不另起分支。resolve 失败返回错误文本,调用方据此把
|
||||
任务置 failed(仓库解析失败绝不标 completed,也不在此重试,
|
||||
避免对不可用的仓库反复 clone 烧时间)。
|
||||
路径,不另起分支。租户模式下检出放在租户目录,task
|
||||
["repo"] 设为容器内路径(挂载目标),而不是宿主检出
|
||||
路径(worker 侧不再用 task["repo"] 当宿主路径)。
|
||||
resolve 失败返回错误文本,调用方据此把任务置 failed
|
||||
(仓库解析失败绝不标 completed,也不在此重试,避免对
|
||||
不可用的仓库反复 clone 烧时间)。
|
||||
"""
|
||||
use_tenant = self.tenant_isolation and \
|
||||
isinstance(self.executor, DockerExecutor)
|
||||
repo_url = self._project_repo_url(task.get("project_id"))
|
||||
if not repo_url:
|
||||
if use_tenant:
|
||||
try:
|
||||
self._tenant_dirs(task.get("project_id"))
|
||||
except (ValueError, RuntimeError) as exc:
|
||||
return "租户目录准备失败: %s" % exc
|
||||
task["repo"] = "/var/lib/codex-app-server/test-repo"
|
||||
return None
|
||||
try:
|
||||
task["repo"] = resolve_repo(
|
||||
repo_url, self.repos_dir, task.get("project_id"),
|
||||
runner=self.repos_runner)
|
||||
if use_tenant:
|
||||
safe = self._tenant_safe_name(task.get("project_id"))
|
||||
resolve_repo(
|
||||
repo_url, os.path.join(self.tenants_root, safe),
|
||||
task.get("project_id"), runner=self.repos_runner)
|
||||
task["repo"] = \
|
||||
"/var/lib/codex-app-server/repos/%s" % safe
|
||||
else:
|
||||
task["repo"] = resolve_repo(
|
||||
repo_url, self.repos_dir, task.get("project_id"),
|
||||
runner=self.repos_runner)
|
||||
except (RepoError, ValueError) as exc:
|
||||
return "仓库解析失败: %s" % exc
|
||||
except Exception as exc:
|
||||
@@ -958,8 +1099,42 @@ class Worker:
|
||||
deadline_seconds=self.task_timeout)
|
||||
else:
|
||||
if is_docker:
|
||||
tenant_mounts = None
|
||||
tenant_network = None
|
||||
if self.tenant_isolation:
|
||||
try:
|
||||
tenant_mounts = self._tenant_mounts(
|
||||
task.get("project_id"))
|
||||
tenant_network = self._tenant_network(
|
||||
task.get("project_id"))
|
||||
except (ValueError, RuntimeError) as exc:
|
||||
# 提前返回会跳过下面的 finally(清容器
|
||||
# 标记),这里先手动清掉,否则 _running
|
||||
# 里永远残留该任务,心跳与退出清理
|
||||
# 都会把它当在跑任务。
|
||||
self._running_clear(task_id)
|
||||
return ({"executor": "docker",
|
||||
"task_id": task_id,
|
||||
"status": "failed", "stdout": "",
|
||||
"returncode": None, "host_status": None,
|
||||
"stderr": ""},
|
||||
"租户隔离准备失败: %s" % exc)
|
||||
# 网络建失败绝不降级默认网络跑容器(降级会
|
||||
# 让隔离悄悄失效);直接让本次尝试 failed,
|
||||
# 调用方按普通失败处理(重试/fallback 照常)。
|
||||
if not self.executor.ensure_network(tenant_network):
|
||||
self._running_clear(task_id)
|
||||
return ({"executor": "docker",
|
||||
"task_id": task_id,
|
||||
"status": "failed", "stdout": "",
|
||||
"returncode": None, "host_status": None,
|
||||
"stderr": ""},
|
||||
"租户 Docker 网络创建失败: %s"
|
||||
% tenant_network)
|
||||
result = self.executor.dispatch_task(
|
||||
task, timeout=effective_docker_timeout)
|
||||
task, timeout=effective_docker_timeout,
|
||||
extra_mounts=tenant_mounts,
|
||||
network=tenant_network)
|
||||
else:
|
||||
result = self.executor.dispatch_task(task)
|
||||
combined = (result.get("stdout", "") or "") + \
|
||||
@@ -968,7 +1143,11 @@ class Worker:
|
||||
_ALREADY_EXISTS_MARKER in combined:
|
||||
if is_docker:
|
||||
result = self.executor.run_task(
|
||||
task, timeout=effective_docker_timeout)
|
||||
task, timeout=effective_docker_timeout,
|
||||
extra_mounts=tenant_mounts
|
||||
if self.tenant_isolation else None,
|
||||
network=tenant_network
|
||||
if self.tenant_isolation else None)
|
||||
else:
|
||||
result = self.executor.run_task(task)
|
||||
except Exception as exc: # runner 异常也如实写回 failed
|
||||
@@ -1585,6 +1764,20 @@ def main(argv=None):
|
||||
"仓库;Docker 执行器下它必须放在"
|
||||
" --docker-extra-mounts 已挂载的目录下,"
|
||||
"否则容器内 task repo 路径不存在")
|
||||
parser.add_argument("--tenant-isolation", action="store_true",
|
||||
help="多租户隔离(仅 Docker 执行器):每个项目用"
|
||||
"独立的 tenants_root/<项目>/work、codex-home、"
|
||||
"test-repo 副本与项目独立 Docker 网络,容器"
|
||||
"内路径不变,宿主挂载源按项目分开")
|
||||
parser.add_argument("--tenants-root",
|
||||
default="/var/lib/codex-app-server/tenants")
|
||||
parser.add_argument("--tenant-template-home",
|
||||
default="/var/lib/codex-app-server/codex-home",
|
||||
help="租户 codex-home 播种模板:只取其中的"
|
||||
"config.toml,绝不复制 thread_history")
|
||||
parser.add_argument("--tenant-template-repo",
|
||||
default="/var/lib/codex-app-server/test-repo",
|
||||
help="租户 test-repo 播种模板(整目录复制)")
|
||||
parser.add_argument("--env-key", default=None)
|
||||
parser.add_argument("--codex-config", default=None,
|
||||
help="含 env_key 行的 codex 配置文件(只读名字)")
|
||||
@@ -1665,7 +1858,11 @@ def main(argv=None):
|
||||
repos_dir=args.repos_dir,
|
||||
heartbeat_file=args.heartbeat_file,
|
||||
worker_id=args.worker_id,
|
||||
capacity=args.worker_capacity)
|
||||
capacity=args.worker_capacity,
|
||||
tenant_isolation=args.tenant_isolation,
|
||||
tenants_root=args.tenants_root,
|
||||
tenant_template_home=args.tenant_template_home,
|
||||
tenant_template_repo=args.tenant_template_repo)
|
||||
# CLI 启动清理兜底(常驻与单次都做):清掉上一轮残留容器。
|
||||
worker.cleanup_stale_containers()
|
||||
# CLI 起 worker 前先建表/迁移:全新库直接起循环不应因缺表而炸。
|
||||
|
||||
@@ -14,7 +14,7 @@ EnvironmentFile=/etc/codex-app-server/env
|
||||
# scaler 会按实例强制追加;--db 只给 scaler,scaler 会补给 worker。
|
||||
# 注意:启用 autoscaler 后必须停掉静态 codex-cloud-worker,
|
||||
# 否则两边同时派任务(scaler 只管理 autoscale-* 实例)。
|
||||
ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos
|
||||
ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos --tenant-isolation --tenants-root /var/lib/codex-app-server/tenants --tenant-template-home /var/lib/codex-app-server/codex-home --tenant-template-repo /var/lib/codex-app-server/test-repo
|
||||
Restart=on-failure
|
||||
RestartSec=10
|
||||
# 停 scaler 时要等它把 worker 优雅排空(drain 60 秒 + 缓冲),
|
||||
|
||||
Reference in new issue
Block a user