134 lines
5.8 KiB
Python
134 lines
5.8 KiB
Python
"""Strict admin-only management; anonymous mode never grants host access."""
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request
|
|
from pydantic import BaseModel
|
|
|
|
from app.gateway.deps import get_current_user_from_request
|
|
|
|
from .controller import PoolUnavailable
|
|
from .docker import DockerError
|
|
from .preflight import shared_storage_configured
|
|
from .settings import PoolSettings
|
|
|
|
router = APIRouter(prefix="/api/system-settings/worker-pool", tags=["worker-pool"])
|
|
|
|
|
|
class DiagnosticsSwitch(BaseModel):
|
|
enabled: bool
|
|
|
|
|
|
@router.get("/diagnostics")
|
|
async def diagnostics_status(request: Request):
|
|
return (await controller(request)).diagnostics.status()
|
|
|
|
|
|
@router.put("/diagnostics")
|
|
async def diagnostics_switch(body: DiagnosticsSwitch, request: Request):
|
|
pool = await controller(request)
|
|
await pool.diagnostics.set_enabled(body.enabled)
|
|
return pool.diagnostics.status()
|
|
|
|
|
|
@router.get("/logs")
|
|
async def container_logs(request: Request, node: str = Query(min_length=1, max_length=128), tail: int = Query(default=200, ge=1, le=1000), minutes: int = Query(default=60, ge=1, le=1440), query: str = Query(default="", max_length=100)):
|
|
pool = await controller(request)
|
|
try:
|
|
return await pool.diagnostics.logs(node, tail, minutes, query)
|
|
except DockerError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
|
|
|
|
@router.get("/log-archives")
|
|
async def log_archives(request: Request):
|
|
return (await controller(request)).diagnostics.archives()
|
|
|
|
|
|
async def controller(request: Request):
|
|
user = await get_current_user_from_request(request)
|
|
if user is None or getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(403, "仅管理员可以管理执行节点")
|
|
from .lifecycle import ensure_controller
|
|
|
|
return ensure_controller(request.app)
|
|
|
|
|
|
@router.get("")
|
|
async def status(request: Request):
|
|
return (await controller(request)).status()
|
|
|
|
|
|
@router.put("")
|
|
async def save(body: PoolSettings, request: Request):
|
|
pool = await controller(request)
|
|
async with pool.lock:
|
|
current = pool.store.load()
|
|
immutable = ("transport", "host", "port", "username", "source_container", "network", "concurrency", "cpu", "memory_mb", "queue_limit", "queue_timeout", "python_executable", "working_dir", "database_pool_size", "postgres_pool_size")
|
|
controller_active = pool.task is not None and not pool.task.done()
|
|
if (pool.nodes or (controller_active and not pool.discovery_complete)) and any(getattr(current, key) != getattr(body, key) for key in immutable):
|
|
raise HTTPException(409, "请先将节点数设为 0，等待所有节点退出后再修改连接或单节点配置")
|
|
config = getattr(request.app.state, "config", None)
|
|
if body.enabled:
|
|
if not shared_storage_configured(config):
|
|
raise HTTPException(409, "启用前必须配置持久化 PostgreSQL 或 SQLite 聊天存储及共享业务库")
|
|
from langgraph.checkpoint.memory import InMemorySaver
|
|
|
|
if isinstance(getattr(request.app.state, "checkpointer", None), InMemorySaver):
|
|
raise HTTPException(409, "当前检查点已降级到内存，不能启用 Worker")
|
|
if "AioSandboxProvider" in str(getattr(getattr(config, "sandbox", None), "use", "")):
|
|
raise HTTPException(409, "当前沙箱需要 Docker 管理权限，在线 Worker 暂不支持；请先配置不依赖 Docker socket 的沙箱")
|
|
if body.desired * body.cpu > body.total_cpu or body.desired * body.memory_mb > body.total_memory_mb:
|
|
raise HTTPException(422, "Worker 总资源超过预算")
|
|
if body.desired * (body.database_pool_size + 2 * body.postgres_pool_size) > body.database_connection_budget:
|
|
raise HTTPException(422, "Worker 数据库连接总额度超过预算")
|
|
identity = ("transport", "host", "port", "username", "host_key_sha256")
|
|
if body.transport == "ssh" and body.password is None and current.password and any(getattr(body, k) != getattr(current, k) for k in identity):
|
|
raise HTTPException(422, "服务器连接身份已改变，请重新填写密码")
|
|
pool.store.save(body)
|
|
try:
|
|
pool.start()
|
|
except RuntimeError as exc:
|
|
pool.store.save(current)
|
|
raise HTTPException(409, str(exc)) from None
|
|
return pool.status()
|
|
|
|
|
|
@router.post("/test")
|
|
async def test_connection(request: Request):
|
|
pool = await controller(request)
|
|
try:
|
|
async with pool.lock:
|
|
await pool.inspect_template(pool.store.load())
|
|
except (DockerError, PoolUnavailable) as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
return {"ok": True, "message": "服务器连接和容器模板检查通过，镜像内 Python 路径已自动校验；节点启动后还会检查聊天存储与运行状态"}
|
|
|
|
|
|
@router.post("/fingerprint")
|
|
async def fingerprint(body: PoolSettings, request: Request):
|
|
import asyncio
|
|
|
|
from .discovery import host_fingerprint
|
|
await controller(request)
|
|
try:
|
|
value = await asyncio.to_thread(host_fingerprint, body.host, body.port)
|
|
except DockerError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
return {"fingerprint": value}
|
|
|
|
|
|
@router.post("/discover")
|
|
async def discover_server(body: PoolSettings, request: Request):
|
|
import asyncio
|
|
|
|
from .discovery import discover
|
|
pool = await controller(request)
|
|
current = pool.store.load()
|
|
if body.password is None:
|
|
identity = ("transport", "host", "port", "username", "host_key_sha256")
|
|
if all(getattr(body, k) == getattr(current, k) for k in identity):
|
|
body = body.model_copy(update={"password": current.password})
|
|
try:
|
|
return await asyncio.to_thread(discover, body, pool.store.raw()["instance"])
|
|
except DockerError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|