feat(servers): 新服务器接入自动 sudo 配置与路径探测

创建/导入/IP 添加成功后后台 onboard 任务;前端注册进度条与 toast。
This commit is contained in:
Nexus Agent
2026-06-12 00:18:55 +08:00
parent 3a4b2ff3eb
commit 17abd62261
17 changed files with 535 additions and 41 deletions
@@ -0,0 +1,41 @@
# 审计 — 新服务器接入自动化
## 范围(文件清单)
| 文件 | 变更 |
|------|------|
| `server/application/services/server_onboarding_service.py` | 调度入口 |
| `server/application/services/server_batch_service.py` | `_run_onboard` |
| `server/application/server_batch_common.py` | `detect_and_save_target_path` |
| `server/api/servers.py` | 创建/导入/IP 添加钩子 |
| `server/api/schemas.py` | `onboarding_job_id` |
| `frontend/src/composables/servers/useServerFormDialog.ts` | 任务注册 |
| `tests/test_server_onboarding.py` | 单测 |
## Step 3 规则扫描
| H | 规则 | 结论 |
|---|------|------|
| H1 | sudoers 仍走白名单模板 | PASS |
| H2 | 创建 API 不因 onboard 失败而 5xx | PASS — 后台 job |
| H3 | 探测失败不阻断推送 | PASS — wwwroot 回退 |
## Closure
| H | 判定 | 依据 |
|---|------|------|
| H1 | PASS | files_sudoers_service |
| H2 | PASS | schedule 独立 try/except |
| H3 | PASS | detect 失败 stdout 提示 |
## DoD
- [x] pytest test_server_onboarding
- [x] changelog 2026-06-11-server-onboarding-automation.md
- [x] 设计 docs/design/specs/2026-06-11-server-onboarding-automation-design.md
## 验证
```bash
pytest tests/test_server_onboarding.py -q
```
@@ -0,0 +1,46 @@
# 审计 — 2026-06-12 新服务器接入自动化(生产部署)
## 范围(文件清单)
| 文件 | 说明 |
|------|------|
| `server/application/services/server_onboarding_service.py` | 调度入口 |
| `server/application/services/server_batch_service.py` | `_run_onboard` |
| `server/application/server_batch_common.py` | `detect_and_save_target_path` |
| `server/api/servers.py` | 创建/导入/IP 添加钩子 |
| `server/api/schemas.py` | `onboarding_job_id` |
| `frontend/src/composables/servers/useServerFormDialog.ts` | 任务注册 |
| `frontend/src/pages/ServersPage.vue` | IP 批量添加注册 |
| `frontend/src/types/api.ts` | 类型 |
| `frontend/src/utils/auditLabels.ts` | batch_server_onboard 标签 |
| `scripts/batch_detect_target_path.py` | 复用 detect_and_save_target_path |
| `tests/test_server_onboarding.py` | 单测 |
## Step 3 规则扫描
| H | 规则 | 结论 |
|---|------|------|
| H1 | sudoers 白名单 | PASS |
| H2 | 创建 API 不 5xx | PASS |
| H3 | 探测失败可推 wwwroot | PASS |
## Closure
| H | 判定 | 依据 |
|---|------|------|
| H1 | PASS | files_sudoers_service |
| H2 | PASS | server_onboarding_service |
| H3 | PASS | _run_onboard |
## DoD
- [x] pytest test_server_onboarding
- [x] changelog 2026-06-12-server-onboarding-automation.md
- [x] 7/7 门控
## 验证
```bash
pytest tests/test_server_onboarding.py -q
bash deploy/pre_deploy_check.sh
```
@@ -0,0 +1,40 @@
# Changelog — 2026-06-11 新服务器接入自动化
## 摘要
创建/导入/凭据轮询添加服务器成功后,后台自动执行 **sudoers 配置(含 rsync** + **目标路径探测**,无需再手动点「一键配置 sudo」和「检测目标路径」。
## 动机
- 非 root 新机推送失败多因未配 sudoers
- 运维希望加机后开箱即用
## 变更
| 文件 | 说明 |
|------|------|
| `server/application/services/server_onboarding_service.py` | 调度 `onboard` 批任务 |
| `server/application/services/server_batch_service.py` | op `onboard` |
| `server/application/server_batch_common.py` | `detect_and_save_target_path` 抽取 |
| `server/api/servers.py` | 创建/导入/IP 添加后触发 |
| `frontend/.../useServerFormDialog.ts` | 进度条 + toast |
| `tests/test_server_onboarding.py` | 单测 |
## 行为
1. **非 root**`install_nexus_files_sudoers`(失败则任务项失败)
2. **未设路径**:探测 workerman.bat;失败仍标记成功,提示可推 `/www/wwwroot`
3. **root**:跳过 sudo,仅探测(若未设路径)
API 响应可选字段:`onboarding_job_id``onboarding_label`
## 迁移 / 重启
需重启 API / 重建镜像。无 DB 迁移。
## 验证
```bash
pytest tests/test_server_onboarding.py -q
# 新建 ubuntu 服务器 → 底部任务栏「新服务器初始化」
```
@@ -0,0 +1,25 @@
# Changelog — 2026-06-12 新服务器接入自动化(生产部署)
## 摘要
部署 `7ba5e21`:创建/导入/IP 添加服务器后自动后台 **onboard**(非 root sudoers + 未设路径探测)。
## 动机
避免新机重复出现推送 Permission denied / 未配 sudoers。
## 变更
同 [2026-06-11-server-onboarding-automation.md](./2026-06-11-server-onboarding-automation.md)。
## 迁移 / 重启
重建 Docker 镜像并重启 API。无 DB 迁移。
## 验证
```bash
pytest tests/test_server_onboarding.py -q
# 生产:新建 ubuntu 子机 → 任务栏「新服务器初始化」
curl -s https://api.synaglobal.vip/health
```
@@ -0,0 +1,19 @@
# 实施 — 新服务器接入自动化
## 文件
| 文件 | 变更 |
|------|------|
| `server/application/server_batch_common.py` | `detect_and_save_target_path` |
| `server/application/services/server_onboarding_service.py` | 调度入口 |
| `server/application/services/server_batch_service.py` | op `onboard` |
| `server/api/servers.py` | 创建/导入后触发 |
| `server/api/schemas.py` | `onboarding_job_id` |
| `frontend/.../useServerFormDialog.ts` | 注册进度条 |
| `tests/test_server_onboarding.py` | 单测 |
## 验证
```bash
pytest tests/test_server_onboarding.py -q
```
@@ -0,0 +1,34 @@
# 设计 — 新服务器接入自动化
## 背景
非 root 子机推送需目标机 `nexus-files` sudoers;未设 `target_path` 时推送回退 `/www/wwwroot`。此前需人工「一键配置 sudo」+「检测目标路径」。
## 目标
创建/导入/凭据轮询添加服务器成功后,**后台自动**:
1. 非 root:安装/补全 `nexus-files` sudoers(含 rsync
2. `target_path` 仍为「未设路径」:SSH 探测 workerman.bat 并写入站点目录
## 方案
新增批量操作 `onboard`,复用现有 `server_batch_service` 后台任务 + WebSocket 进度。
| 入口 | 触发 |
|------|------|
| POST `/servers/` | 单台 → job |
| CSV import | 批量 created_ids → 一个 job |
| 凭据轮询 / add-by-ip / pending retry | `_create_server_from_poll` 内触发 |
root SSH:跳过 sudo 步骤;已有具体 `target_path`:跳过探测。
## 非目标
- 不自动安装 Agent
- 不因 onboarding 失败阻断创建 API(仅记录 batch job 失败项)
## 验收
- 新建 ubuntu 服务器 → 返回 `onboarding_job_id`,任务含 sudo + detect
- pytest 覆盖 onboard 编排逻辑
@@ -1,6 +1,8 @@
import { ref, computed } from 'vue' import { ref, computed } from 'vue'
import { http } from '@/api' import { http } from '@/api'
import { useSnackbar } from '@/composables/useSnackbar' import { useSnackbar } from '@/composables/useSnackbar'
import { registerServerBatchJob } from '@/composables/useScriptExecutionQueue'
import { showScriptSubmitToast } from '@/composables/useScriptSubmitToast'
import { formatApiError } from '@/utils/apiError' import { formatApiError } from '@/utils/apiError'
import type { FilesElevationMode, ServerApiItem } from '@/types/api' import type { FilesElevationMode, ServerApiItem } from '@/types/api'
@@ -229,7 +231,21 @@ export function useServerFormDialog(onSaved: () => void) {
if (isEditing.value && editingId.value != null) { if (isEditing.value && editingId.value != null) {
await http.put(`/servers/${editingId.value}`, payload) await http.put(`/servers/${editingId.value}`, payload)
} else { } else {
await http.post('/servers/', payload) const created = await http.post<ServerApiItem & {
onboarding_job_id?: number
onboarding_label?: string
}>('/servers/', payload)
if (created.onboarding_job_id) {
registerServerBatchJob(
created.onboarding_job_id,
created.onboarding_label || '新服务器初始化',
1,
'onboard',
'0/1',
'running',
)
showScriptSubmitToast('新服务器初始化已在后台执行(sudo + 路径探测)')
}
} }
close() close()
onSaved() onSaved()
+20
View File
@@ -727,6 +727,23 @@ async function submitServerBatchJob(
return res return res
} }
function registerOnboardingJob(
jobId: number | undefined,
label: string | undefined,
total: number,
) {
if (!jobId) return
registerServerBatchJob(
jobId,
label || '新服务器初始化',
total,
'onboard',
`0/${total}`,
'running',
)
showScriptSubmitToast('新服务器初始化已在后台执行(sudo + 路径探测)')
}
onMounted(() => { onMounted(() => {
offBatchComplete = onScriptExecutionComplete((alert) => { offBatchComplete = onScriptExecutionComplete((alert) => {
if (alert.taskKind !== 'server_batch') return if (alert.taskKind !== 'server_batch') return
@@ -1173,6 +1190,7 @@ async function submitQuickAdd() {
const res = await http.post<AddByIpResponse>('/servers/add-by-ip', body) const res = await http.post<AddByIpResponse>('/servers/add-by-ip', body)
if (res.success && res.server) { if (res.success && res.server) {
showQuickAdd.value = false showQuickAdd.value = false
registerOnboardingJob(res.onboarding_job_id, res.onboarding_label, 1)
snackbar(`添加成功:${res.matched_preset} (${res.matched_username})`) snackbar(`添加成功:${res.matched_preset} (${res.matched_username})`)
loadServers() loadServers()
loadStats() loadStats()
@@ -1193,6 +1211,7 @@ async function retryPending(item: PendingServerItem) {
try { try {
const res = await http.post<AddByIpResponse>(`/servers/pending/${item.id}/retry`, {}) const res = await http.post<AddByIpResponse>(`/servers/pending/${item.id}/retry`, {})
if (res.success && res.server) { if (res.success && res.server) {
registerOnboardingJob(res.onboarding_job_id, res.onboarding_label, 1)
snackbar(`重试成功:${res.matched_preset} (${res.matched_username})`) snackbar(`重试成功:${res.matched_preset} (${res.matched_username})`)
loadServers() loadServers()
loadStats() loadStats()
@@ -1705,6 +1724,7 @@ async function submitBatchAdd() {
}) })
batchAddResult.value = res batchAddResult.value = res
if (res.created > 0) { if (res.created > 0) {
registerOnboardingJob(res.onboarding_job_id, '新服务器初始化', res.created)
loadServers() loadServers()
loadStats() loadStats()
} }
+3
View File
@@ -266,6 +266,8 @@ export interface AddByIpResponse {
message?: string message?: string
matched_preset?: string matched_preset?: string
matched_username?: string matched_username?: string
onboarding_job_id?: number
onboarding_label?: string
} }
/** Item from POST /api/servers/add-by-ips-batch */ /** Item from POST /api/servers/add-by-ips-batch */
@@ -289,6 +291,7 @@ export interface AddByIpsBatchResult {
input_lines?: number input_lines?: number
duplicates_removed?: number duplicates_removed?: number
items: AddByIpsBatchItemResult[] items: AddByIpsBatchItemResult[]
onboarding_job_id?: number
} }
/** Alert history item from /api/alert-history/ */ /** Alert history item from /api/alert-history/ */
+1
View File
@@ -39,6 +39,7 @@ const ACTION_LABELS: Record<string, string> = {
batch_upgrade_agent: '批量升级 Agent', batch_upgrade_agent: '批量升级 Agent',
batch_uninstall_agent: '批量卸载 Agent', batch_uninstall_agent: '批量卸载 Agent',
batch_detect_path: '批量检测路径', batch_detect_path: '批量检测路径',
batch_server_onboard: '新服务器初始化',
batch_health_check: '批量健康检查', batch_health_check: '批量健康检查',
batch_update_category: '批量修改分类', batch_update_category: '批量修改分类',
+7 -20
View File
@@ -67,38 +67,25 @@ async def _fetch_servers(
async def _detect_one_server(server: Any) -> dict[str, Any]: async def _detect_one_server(server: Any) -> dict[str, Any]:
from server.application.server_batch_common import detect_and_save_target_path
from server.application.services.server_batch_service import result_item_dict from server.application.services.server_batch_service import result_item_dict
from server.application.server_batch_common import sudo_wrap, update_server_target_path
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
from server.utils.posix_paths import posix_dirname
sid = server.id sid = server.id
label = server.name or str(sid) label = server.name or str(sid)
try: try:
ssh_user = (server.username or "root").strip() outcome = await detect_and_save_target_path(server)
cmd = "find /www/wwwroot -iname 'workerman.bat' -type f -print -quit 2>/dev/null" if outcome.get("success"):
cmd = sudo_wrap(cmd, ssh_user)
r = await exec_ssh_command(server, cmd, timeout=30)
if r["exit_code"] != 0 and not r.get("stdout", "").strip():
return result_item_dict( return result_item_dict(
server_id=sid, server_id=sid,
server_name=label, server_name=label,
success=False, success=True,
error=f"搜索失败 (exit {r['exit_code']})", stdout=str(outcome.get("stdout") or ""),
) )
found = r["stdout"].strip()
if not found:
return result_item_dict(
server_id=sid, server_name=label, success=False, error="未找到 workerman.bat"
)
first_match = found.split("\n")[0].strip()
target_dir = posix_dirname(first_match)
await update_server_target_path(sid, target_dir)
return result_item_dict( return result_item_dict(
server_id=sid, server_id=sid,
server_name=label, server_name=label,
success=True, success=False,
stdout=f"target_path → {target_dir}", error=str(outcome.get("error") or "探测失败"),
) )
except Exception as e: except Exception as e:
return result_item_dict(server_id=sid, server_name=label, success=False, error=str(e)[:300]) return result_item_dict(server_id=sid, server_name=label, success=False, error=str(e)[:300])
+2
View File
@@ -130,6 +130,7 @@ class ServerImportResult(BaseModel):
failed: int = 0 failed: int = 0
errors: List[dict] = [] # [{row, name, domain, error}] errors: List[dict] = [] # [{row, name, domain, error}]
created_ids: List[int] = [] created_ids: List[int] = []
onboarding_job_id: Optional[int] = None
class BatchAgentAction(BaseModel): class BatchAgentAction(BaseModel):
@@ -772,6 +773,7 @@ class AddByIpsBatchResult(BaseModel):
input_lines: int = Field(0, description="非空输入行数(去重前)") input_lines: int = Field(0, description="非空输入行数(去重前)")
duplicates_removed: int = Field(0, description="输入中重复行已忽略的数量") duplicates_removed: int = Field(0, description="输入中重复行已忽略的数量")
items: List[AddByIpsBatchItemResult] = [] items: List[AddByIpsBatchItemResult] = []
onboarding_job_id: Optional[int] = None
# ── Platform / Node ── # ── Platform / Node ──
+41 -3
View File
@@ -49,6 +49,19 @@ router = APIRouter(prefix="/api/servers", tags=["servers"])
REDIS_KEY_PREFIX = "heartbeat:" REDIS_KEY_PREFIX = "heartbeat:"
async def _schedule_server_onboarding(server_ids: list[int], operator: str) -> int | None:
"""Fire-and-forget batch job: sudoers + detect-path for newly created servers."""
if not server_ids:
return None
from server.application.services.server_onboarding_service import schedule_server_onboarding
job = await schedule_server_onboarding(server_ids, operator=operator or "admin")
if not job:
return None
job_id = job.get("job_id")
return int(job_id) if job_id is not None else None
def _server_has_agent_monitoring(server: Server) -> bool: def _server_has_agent_monitoring(server: Server) -> bool:
"""True when this server is expected to run Nexus Agent (key or version on file).""" """True when this server is expected to run Nexus Agent (key or version on file)."""
return bool((server.agent_api_key or "").strip() or (server.agent_version or "").strip()) return bool((server.agent_api_key or "").strip() or (server.agent_version or "").strip())
@@ -590,6 +603,10 @@ async def import_servers(
result.total = result.created + result.skipped + result.failed result.total = result.created + result.skipped + result.failed
onboarding_job_id = await _schedule_server_onboarding(result.created_ids, admin.username)
if onboarding_job_id is not None:
result.onboarding_job_id = onboarding_job_id
# Audit log # Audit log
ip_address = request.client.host if request.client else "" ip_address = request.client.host if request.client else ""
audit_repo = AuditLogRepositoryImpl(db) audit_repo = AuditLogRepositoryImpl(db)
@@ -961,12 +978,19 @@ async def add_server_by_ip(
raise HTTPException(409, f"服务器 {item.domain} 已存在于列表中") raise HTTPException(409, f"服务器 {item.domain} 已存在于列表中")
if item.status == "created": if item.status == "created":
return { onboarding_job_id = None
if item.server_id is not None:
onboarding_job_id = await _schedule_server_onboarding([item.server_id], admin.username)
body: dict = {
"success": True, "success": True,
"server": {"id": item.server_id, "domain": item.domain}, "server": {"id": item.server_id, "domain": item.domain},
"matched_preset": item.matched_preset, "matched_preset": item.matched_preset,
"matched_username": item.matched_username, "matched_username": item.matched_username,
} }
if onboarding_job_id is not None:
body["onboarding_job_id"] = onboarding_job_id
body["onboarding_label"] = "新服务器初始化"
return body
return { return {
"success": False, "success": False,
@@ -1022,6 +1046,9 @@ async def add_servers_by_ips_batch(
else: else:
skipped += 1 skipped += 1
onboarding_ids = [i.server_id for i in items if i.status == "created" and i.server_id is not None]
onboarding_job_id = await _schedule_server_onboarding(onboarding_ids, admin.username)
ip_address = request.client.host if request.client else "" ip_address = request.client.host if request.client else ""
audit_repo = AuditLogRepositoryImpl(db) audit_repo = AuditLogRepositoryImpl(db)
await audit_repo.create(AuditLog( await audit_repo.create(AuditLog(
@@ -1044,6 +1071,7 @@ async def add_servers_by_ips_batch(
input_lines=parsed.input_lines, input_lines=parsed.input_lines,
duplicates_removed=parsed.duplicates_removed, duplicates_removed=parsed.duplicates_removed,
items=items, items=items,
onboarding_job_id=onboarding_job_id,
) )
@@ -1090,12 +1118,17 @@ async def retry_pending_server(
request=request, request=request,
) )
await pending_repo.delete(pending_id) await pending_repo.delete(pending_id)
return { onboarding_job_id = await _schedule_server_onboarding([created.id], admin.username)
body: dict = {
"success": True, "success": True,
"server": _server_to_dict(created), "server": _server_to_dict(created),
"matched_preset": poll.match.preset_name, "matched_preset": poll.match.preset_name,
"matched_username": poll.match.username, "matched_username": poll.match.username,
} }
if onboarding_job_id is not None:
body["onboarding_job_id"] = onboarding_job_id
body["onboarding_label"] = "新服务器初始化"
return body
err_text = format_poll_errors(poll.errors) err_text = format_poll_errors(poll.errors)
pending.attempts = (pending.attempts or 0) + 1 pending.attempts = (pending.attempts or 0) + 1
@@ -1346,7 +1379,12 @@ async def create_server(
ip_address=ip_address, ip_address=ip_address,
)) ))
return _server_to_dict(created) resp = _server_to_dict(created)
onboarding_job_id = await _schedule_server_onboarding([created.id], admin.username)
if onboarding_job_id is not None:
resp["onboarding_job_id"] = onboarding_job_id
resp["onboarding_label"] = "新服务器初始化"
return resp
@router.put("/{id}", response_model=dict) @router.put("/{id}", response_model=dict)
+23
View File
@@ -144,6 +144,29 @@ async def update_server_target_path(sid: int, target_dir: str) -> None:
await session.commit() await session.commit()
async def detect_and_save_target_path(server: Server, *, timeout: int = 30) -> dict[str, object]:
"""Find workerman.bat under /www/wwwroot and persist parent dir as target_path."""
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
sid = server.id
ssh_user = (server.username or "root").strip() or "root"
cmd = "find /www/wwwroot -iname 'workerman.bat' -type f -print -quit 2>/dev/null"
cmd = sudo_wrap(cmd, ssh_user)
r = await exec_ssh_command(server, cmd, timeout=timeout)
if r["exit_code"] != 0 and not r.get("stdout", "").strip():
return {
"success": False,
"error": f"搜索失败 (exit {r['exit_code']})",
}
found = (r.get("stdout") or "").strip()
if not found:
return {"success": False, "error": "未找到 workerman.bat"}
first_match = found.split("\n")[0].strip()
target_dir = posix_dirname(first_match)
await update_server_target_path(sid, target_dir)
return {"success": True, "target_dir": target_dir, "stdout": f"target_path → {target_dir}"}
async def mark_agent_uninstalled(sid: int) -> None: async def mark_agent_uninstalled(sid: int) -> None:
from server.infrastructure.database.session import AsyncSessionLocal from server.infrastructure.database.session import AsyncSessionLocal
from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.server_repo import ServerRepositoryImpl
@@ -10,14 +10,13 @@ from typing import Any, Optional
from server.application.server_batch_common import ( from server.application.server_batch_common import (
agent_install_skip_result, agent_install_skip_result,
batch_server_display_name, batch_server_display_name,
detect_and_save_target_path,
ensure_agent_api_key, ensure_agent_api_key,
install_error_msg, install_error_msg,
mark_agent_uninstalled, mark_agent_uninstalled,
posix_dirname,
prefetch_batch_context, prefetch_batch_context,
result_item_dict, result_item_dict,
sudo_wrap, sudo_wrap,
update_server_target_path,
) )
from server.utils.agent_version import get_central_agent_version from server.utils.agent_version import get_central_agent_version
from server.config import settings from server.config import settings
@@ -38,6 +37,7 @@ OP_LABELS = {
"category": "批量修改分类", "category": "批量修改分类",
"health-check": "健康检查", "health-check": "健康检查",
"detect-path": "自动检测目标路径", "detect-path": "自动检测目标路径",
"onboard": "新服务器初始化",
"install-agent": "批量安装 Agent", "install-agent": "批量安装 Agent",
"upgrade-agent": "批量升级 Agent", "upgrade-agent": "批量升级 Agent",
"uninstall-agent": "批量卸载 Agent", "uninstall-agent": "批量卸载 Agent",
@@ -47,6 +47,7 @@ AUDIT_ACTIONS = {
"category": "batch_update_category", "category": "batch_update_category",
"health-check": "batch_health_check", "health-check": "batch_health_check",
"detect-path": "batch_detect_path", "detect-path": "batch_detect_path",
"onboard": "batch_server_onboard",
"install-agent": "batch_install_agent", "install-agent": "batch_install_agent",
"upgrade-agent": "batch_upgrade_agent", "upgrade-agent": "batch_upgrade_agent",
"uninstall-agent": "batch_uninstall_agent", "uninstall-agent": "batch_uninstall_agent",
@@ -325,6 +326,8 @@ async def _run_background(job_id: int) -> None:
await _run_health_check(live) await _run_health_check(live)
elif op == "detect-path": elif op == "detect-path":
await _run_detect_path(live) await _run_detect_path(live)
elif op == "onboard":
await _run_onboard(live)
elif op == "install-agent": elif op == "install-agent":
await _run_install_agent(live) await _run_install_agent(live)
elif op == "upgrade-agent": elif op == "upgrade-agent":
@@ -373,6 +376,7 @@ async def _write_audit(live: dict[str, Any]) -> None:
"category": f"批量修改分类 更新{stats['success']}", "category": f"批量修改分类 更新{stats['success']}",
"health-check": f"批量健康检查: {stats['success']}/{stats['total']} 在线", "health-check": f"批量健康检查: {stats['success']}/{stats['total']} 在线",
"detect-path": f"批量检测路径: {stats['success']}/{stats['total']} 成功", "detect-path": f"批量检测路径: {stats['success']}/{stats['total']} 成功",
"onboard": f"新服务器初始化: {stats['success']}/{stats['total']} 成功",
"install-agent": f"批量安装Agent: {stats['success']}/{stats['total']} 成功", "install-agent": f"批量安装Agent: {stats['success']}/{stats['total']} 成功",
"upgrade-agent": f"批量升级Agent: {stats['success']}/{stats['total']} 成功", "upgrade-agent": f"批量升级Agent: {stats['success']}/{stats['total']} 成功",
"uninstall-agent": f"批量卸载Agent: {stats['success']}/{stats['total']} 成功", "uninstall-agent": f"批量卸载Agent: {stats['success']}/{stats['total']} 成功",
@@ -497,24 +501,19 @@ async def _run_detect_path(live: dict[str, Any]) -> None:
if not server: if not server:
return result_item_dict(server_id=sid, server_name=label, success=False, error="服务器不存在") return result_item_dict(server_id=sid, server_name=label, success=False, error="服务器不存在")
try: try:
ssh_user = (server.username or "root").strip() outcome = await detect_and_save_target_path(server)
cmd = "find /www/wwwroot -iname 'workerman.bat' -type f -print -quit 2>/dev/null" if outcome.get("success"):
cmd = sudo_wrap(cmd, ssh_user)
r = await exec_ssh_command(server, cmd, timeout=30)
if r["exit_code"] != 0 and not r.get("stdout", "").strip():
return result_item_dict( return result_item_dict(
server_id=sid, server_name=label, success=False, server_id=sid,
error=f"搜索失败 (exit {r['exit_code']})", server_name=label,
success=True,
stdout=str(outcome.get("stdout") or ""),
) )
found = r["stdout"].strip()
if not found:
return result_item_dict(server_id=sid, server_name=label, success=False, error="未找到 workerman.bat")
first_match = found.split("\n")[0].strip()
target_dir = posix_dirname(first_match)
await update_server_target_path(sid, target_dir)
return result_item_dict( return result_item_dict(
server_id=sid, server_name=label, success=True, server_id=sid,
stdout=f"target_path → {target_dir}", server_name=label,
success=False,
error=str(outcome.get("error") or "探测失败"),
) )
except Exception as e: except Exception as e:
return result_item_dict(server_id=sid, server_name=label, success=False, error=str(e)[:300]) return result_item_dict(server_id=sid, server_name=label, success=False, error=str(e)[:300])
@@ -522,6 +521,61 @@ async def _run_detect_path(live: dict[str, Any]) -> None:
await _run_with_semaphore(live, handler) await _run_with_semaphore(live, handler)
async def _run_onboard(live: dict[str, Any]) -> None:
from server.application.services.files_sudoers_service import install_nexus_files_sudoers
from server.utils.posix_paths import is_unset_target_path
async def handler(sid: int, server_map: dict[int, Server], labels: dict[int, str], _live) -> dict:
server = server_map.get(sid)
label = labels[sid]
if not server:
return result_item_dict(server_id=sid, server_name=label, success=False, error="服务器不存在")
steps: list[str] = []
try:
ssh_user = (server.username or "root").strip() or "root"
if ssh_user == "root":
steps.append("sudo: skip(root)")
else:
sudo_outcome = await install_nexus_files_sudoers(server)
action = sudo_outcome.get("action") or "unknown"
if not sudo_outcome.get("ok"):
detail = sudo_outcome.get("detail") or action
return result_item_dict(
server_id=sid,
server_name=label,
success=False,
error=f"sudo 配置失败: {detail}",
stdout="; ".join(steps),
)
steps.append(f"sudo: {action}")
if is_unset_target_path(server.target_path):
detect_outcome = await detect_and_save_target_path(server)
if detect_outcome.get("success"):
steps.append(str(detect_outcome.get("stdout") or "path: ok"))
else:
steps.append(f"path: {detect_outcome.get('error') or 'fail'}(推送仍可用 /www/wwwroot")
else:
steps.append("path: already_set")
return result_item_dict(
server_id=sid,
server_name=label,
success=True,
stdout="; ".join(steps),
)
except Exception as e:
return result_item_dict(
server_id=sid,
server_name=label,
success=False,
error=str(e)[:300],
stdout="; ".join(steps),
)
await _run_with_semaphore(live, handler)
async def _run_install_agent(live: dict[str, Any]) -> None: async def _run_install_agent(live: dict[str, Any]) -> None:
base_url = (settings.API_BASE_URL or "").strip().rstrip("/") base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
if not base_url: if not base_url:
@@ -0,0 +1,30 @@
"""Schedule post-create server onboarding (sudoers + target path detect)."""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger("nexus.server_onboarding")
async def schedule_server_onboarding(
server_ids: list[int],
*,
operator: str,
) -> dict[str, Any] | None:
"""Start background batch job ``onboard`` for *server_ids*; returns job dict or None."""
ids = [int(x) for x in server_ids if x]
if not ids:
return None
from server.application.services.server_batch_service import start_batch_job
try:
return await start_batch_job(
op="onboard",
server_ids=ids,
operator=operator or "admin",
)
except Exception as exc:
logger.warning("schedule_server_onboarding failed ids=%s: %s", ids, exc)
return None
+115
View File
@@ -0,0 +1,115 @@
"""Tests for post-create server onboarding automation."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from server.utils.posix_paths import is_unset_target_path
@pytest.mark.parametrize(
"target_path,expect_detect",
[
("", True),
("/www/wwwroot", True),
("/www/wwwroot/site", False),
],
)
def test_onboard_skips_detect_when_path_set(target_path, expect_detect):
assert is_unset_target_path(target_path) is expect_detect
@pytest.mark.asyncio
async def test_schedule_server_onboarding_delegates_to_batch():
from server.application.services.server_onboarding_service import schedule_server_onboarding
with patch(
"server.application.services.server_batch_service.start_batch_job",
new_callable=AsyncMock,
return_value={"job_id": 99, "status": "running", "total": 1},
) as start:
job = await schedule_server_onboarding([12], operator="admin")
start.assert_awaited_once_with(op="onboard", server_ids=[12], operator="admin")
assert job["job_id"] == 99
@pytest.mark.asyncio
async def test_run_onboard_non_root_success():
from server.application.services.server_batch_service import _run_onboard
from server.domain.models import Server
from server.infrastructure.redis import server_batch_store as sbs
server = Server(id=1, name="t", domain="1.2.3.4", username="ubuntu", target_path="")
live = sbs.init_live(
job_id=1,
op="onboard",
label="新服务器初始化",
server_ids=[1],
operator="admin",
)
async def invoke(_live, handler):
item = await handler(1, {1: server}, {1: "t"}, _live)
sbs.record_result(_live, item)
with patch(
"server.application.services.server_batch_service._run_with_semaphore",
new_callable=AsyncMock,
side_effect=invoke,
):
with patch(
"server.application.services.files_sudoers_service.install_nexus_files_sudoers",
new_callable=AsyncMock,
return_value={"ok": True, "action": "already_ok"},
):
with patch(
"server.application.server_batch_common.detect_and_save_target_path",
new_callable=AsyncMock,
return_value={"success": True, "stdout": "target_path → /www/wwwroot/shop"},
):
await _run_onboard(live)
assert live["results"]["1"]["success"] is True
assert "sudo" in live["results"]["1"]["stdout"]
@pytest.mark.asyncio
async def test_run_onboard_detect_fail_still_success():
from server.application.services.server_batch_service import _run_onboard
from server.domain.models import Server
from server.infrastructure.redis import server_batch_store as sbs
server = Server(id=3, name="u", domain="3.4.5.6", username="ubuntu", target_path="")
live = sbs.init_live(
job_id=3,
op="onboard",
label="新服务器初始化",
server_ids=[3],
operator="admin",
)
async def invoke(_live, handler):
item = await handler(3, {3: server}, {3: "u"}, _live)
sbs.record_result(_live, item)
with patch(
"server.application.services.server_batch_service._run_with_semaphore",
new_callable=AsyncMock,
side_effect=invoke,
):
with patch(
"server.application.services.files_sudoers_service.install_nexus_files_sudoers",
new_callable=AsyncMock,
return_value={"ok": True, "action": "patched"},
):
with patch(
"server.application.server_batch_common.detect_and_save_target_path",
new_callable=AsyncMock,
return_value={"success": False, "error": "未找到 workerman.bat"},
):
await _run_onboard(live)
assert live["results"]["3"]["success"] is True
assert "www/wwwroot" in live["results"]["3"]["stdout"]