feat(watch): 监测槽开关、8h/24h、到期暂停与探针修复

支持随时暂停/恢复实时监测并保留槽位;TTL 到期自动暂停不占删槽;修复到期竞态导致空槽与唯一约束冲突;探针 parse_error 与中文错误;移除失败 snackbar。
This commit is contained in:
Nexus Agent
2026-06-12 03:47:56 +08:00
parent 7bb85aac41
commit 32a1885f0d
26 changed files with 1437 additions and 126 deletions
@@ -0,0 +1,117 @@
# 审计 — 监测槽开关 / 8h·24h / 暂停占槽 / 探针中文错误
**日期**2026-06-11
**Changelog**:
- `docs/changelog/2026-06-11-watch-slot-monitoring-toggle.md`
- `docs/changelog/2026-06-11-watch-probe-parse-error-fix.md`
## 范围(未提交工作区,+799 / −124 行)
| 层级 | 文件 |
|------|------|
| 模型/迁移 | `server/domain/models/__init__.py``monitoring_enabled``ttl_hours` |
| | `server/infrastructure/database/migrations.py` |
| Repository | `server/infrastructure/database/watch_repo.py` |
| Service | `server/application/services/watch_service.py` |
| API | `server/api/watch.py`PATCH monitoring / ttl |
| 探针 | `server/background/watch_probe_runner.py` |
| | `server/infrastructure/ssh/remote_probe.py` |
| 工具 | `server/utils/watch_metrics.py` |
| | `server/utils/watch_state.py`**新增** Redis 清理) |
| | `server/utils/watch_probe_errors.py`**新增** 中文错误) |
| 前端 | `frontend/src/composables/useWatchPins.ts` |
| | `frontend/src/components/watch/WatchSlotCard.vue` |
| | `frontend/src/components/watch/WatchSlotRow.vue` |
| | `frontend/src/components/watch/WatchProbeRecordsTable.vue` |
| | `frontend/src/utils/watchFormat.ts` |
| | `frontend/src/api/index.ts``http.patch` |
| 测试 | `tests/test_watch_pins.py` |
| | `tests/test_watch_metrics.py` |
| | `tests/test_watch_probe_errors.py`**新增** |
## 功能摘要
1. **实时监测开关**`monitoring_enabled`;关则停 5s 探针,槽位保留
2. **8h / 24h**`ttl_hours` + `PATCH /pins/{slot}/ttl`;默认 8h,显式切 24h
3. **暂停永久占槽**`monitoring_enabled=false` 时不参与 TTL 过期删除
4. **时长到期**:自动暂停(等同关开关),结束 session,**不删 pin**
5. **恢复 Agent 60s**:无活跃监测时清理 `watch:live/proc/last`Agent 心跳逻辑未改
6. **parse_error 修复**Agent 有 CPU/内存时 SSH 失败不拖垮整轮;SSH 脚本加固
7. **错误中文**:后端写入 + 前端 `formatProbeError` 兼容历史英文
## Step 3 规则扫描
| H | 规则 | 结论 |
|---|------|------|
| H1 | 鉴权 / IDOR | **PASS** — 新 PATCH 均 `get_current_admin`repo 带 `admin_id` |
| H2 | 无静默吞错 | **PASS** — 探针失败仍落库;Redis 清理失败会抛错(生产有 Redis) |
| H3 | 无新密钥 | **PASS** |
| H4 | CUD 审计 | **PASS**`watch_pin_pause` / `watch_pin_resume` / `watch_pin_ttl` / remove |
| H5 | 输入边界 | **PASS**`Literal[8,24]``slot_index` 03 |
| H6 | 迁移 | **PASS** — ALTER + duplicate column 容错 |
| H7 | 产品风险 | **RISK 接受** — 4 槽均可永久暂停占用,需手动 X 释放(用户明确要求) |
## 逐模块走读
### 后端
| 模块 | 结论 | 备注 |
|------|------|------|
| `watch_repo._pin_still_occupied` | PASS | 暂停不占 TTL;开启监测才看过期 |
| `expire_due_pins` | PASS | 改为暂停 + end session,不 delete pin |
| `set_monitoring_enabled` + `_ensure_open_session` | PASS | TTL 到期后再开自动新 session |
| `_restore_idle_agent_watch` | PASS | 全局无监测才清 Redis |
| `replace_slot` | PASS | 保留槽位原 `ttl_hours`(审查项已修) |
| `probe_server_metrics` | PASS | Agent 基础指标优先;SSH 仅补 IO |
| `watch_probe_error_zh` | PASS | 与前端映射一致 |
### 前端
| 模块 | 结论 | 备注 |
|------|------|------|
| `WatchSlotCard` | PASS | 开关 + 8h/24h + 暂停态 UI |
| `tickCountdown` | PASS | 归零 `refreshPins`,不再误置 `empty` |
| `formatProbeError` | PASS | 状态/错误/来源中文化 |
| `WatchProbeRecordsTable` | PASS | 筛选器与列中文 |
## 审计中发现并修复
| 项 | 严重度 | 处理 |
|----|--------|------|
| `mock_watch_redis` 未 patch `watch_state.get_redis` | **P1** | 已修:`remove_slot` / `set_slot_monitoring` 单测 2 失败 → 19 passed |
## 遗留建议(非阻塞)
| 项 | 说明 |
|----|------|
| 设计文档 | 开关/暂停语义变更较大,建议补 `docs/design/specs/` 短文(当前仅 changelog |
| HTTP 集成测试 | PATCH monitoring/ttl 的 404/422 路径未覆盖 |
| TTL 到期审计 | 后台 `expire_due_pins` 无 audit_log(可接受,session.end_reason=expired 可查) |
| PATCH 404 语义 | 「槽位无效」与「槽位为空」均 404,与 POST 422 不一致 |
## Closure
| H | 判定 | 依据 |
|---|------|------|
| H1H6 | **PASS** | 鉴权、审计、迁移、探针落库均符合铁律 |
| H7 | **RISK 接受** | 用户要求暂停永久占槽 |
## DoD
- [x] `.venv/bin/pytest tests/test_watch_metrics.py tests/test_watch_pins.py tests/test_watch_probe_errors.py`**19 passed**
- [x] `cd frontend && npm run type-check`
- [x] 审计修复:test fixture patch `watch_state.get_redis`
- [ ] `bash scripts/local_verify.sh` / 门控 7 道(部署前)
- [ ] 生产浏览器:开关暂停、24h 切换、到期自动暂停、中文错误展示
## 验证命令
```bash
.venv/bin/pytest tests/test_watch_metrics.py tests/test_watch_pins.py tests/test_watch_probe_errors.py -q
cd frontend && npm run type-check
bash deploy/pre_deploy_check.sh
```
## 合并建议
**Approve with notes** — 功能完整、测试绿、安全与审计达标;部署前跑门控 + 浏览器终验。遗留 HTTP 集成测试与设计文档可跟进 PR。
@@ -0,0 +1,67 @@
# 审计 — 监测槽开关 / TTL / 到期竞态 / 探针中文 / 移除提示
**日期**2026-06-12
**Changelog**:
- `docs/changelog/2026-06-11-watch-slot-monitoring-toggle.md`
- `docs/changelog/2026-06-11-watch-probe-parse-error-fix.md`
- `docs/changelog/2026-06-12-watch-ttl-expiry-race-fix.md`
## 范围(文件清单)
| 层级 | 文件 |
|------|------|
| 模型/迁移 | `server/domain/models/__init__.py` |
| | `server/infrastructure/database/migrations.py` |
| Repository | `server/infrastructure/database/watch_repo.py` |
| Service | `server/application/services/watch_service.py` |
| API | `server/api/watch.py` |
| 探针 | `server/background/watch_probe_runner.py` |
| | `server/infrastructure/ssh/remote_probe.py` |
| 工具 | `server/utils/watch_metrics.py` |
| | `server/utils/watch_state.py` |
| | `server/utils/watch_probe_errors.py` |
| 前端 | `frontend/src/api/index.ts` |
| | `frontend/src/composables/useWatchPins.ts` |
| | `frontend/src/components/watch/WatchSlotCard.vue` |
| | `frontend/src/components/watch/WatchSlotRow.vue` |
| | `frontend/src/components/watch/WatchProbeRecordsTable.vue` |
| | `frontend/src/utils/watchFormat.ts` |
| 测试 | `tests/test_watch_pins.py` |
| | `tests/test_watch_metrics.py` |
| | `tests/test_watch_probe_errors.py` |
## Step 3 规则扫描
| H | 规则 | 结论 |
|---|------|------|
| H1 | 鉴权 / IDOR | PASS — PATCH/DELETE 均 `get_current_admin` + admin_id 隔离 |
| H2 | 无静默吞错 | PASS — 探针失败落库;前端 remove/monitoring/ttl 均有 snackbar |
| H3 | 无新密钥 | PASS |
| H4 | CUD 审计 | PASS — pause/resume/ttl/remove |
| H5 | 输入边界 | PASS — `Literal[8,24]`、slot 03 |
| H6 | 迁移 | PASS — monitoring_enabled + ttl_hours ALTER |
| H7 | TTL 竞态 | PASS — `_finalize_due_pins` 读/写路径同步到期 |
## Closure
| H | 判定 | 依据 |
|---|------|------|
| H1H6 | PASS | 鉴权、审计、迁移、探针落库 |
| H7 | PASS | 到期 ghost pin 不再显示空槽 |
## DoD
- [x] `pytest tests/test_watch_*.py` — 21 passed
- [x] `npm run type-check`
- [x] P0 TTL 竞态修复 + P2 onRemove snackbar
- [ ] `bash deploy/pre_deploy_check.sh`
- [ ] 生产浏览器终验
## 验证命令
```bash
.venv/bin/pytest tests/test_watch_*.py -q
cd frontend && npm run type-check
bash scripts/local_verify.sh
bash deploy/pre_deploy_check.sh
```
@@ -0,0 +1,41 @@
# 监测槽 parse_error 修复
**日期**: 2026-06-11
## 摘要
修复监测槽显示 `parse_error` 的常见路径:Agent 有心跳但缺 IO 时 SSH 补采失败导致整轮探针失败。
## 根因
- `redis_sample_has_io` 为 false 时强制走 SSHSSH JSON 解析失败且无完整 Redis IO 时返回 `parse_error`
- 已装 Agent 的机器通常有 CPU/内存心跳,不应因 SSH 失败而整轮失败
- SSH 脚本在 `getloadavg`/`net_io_counters` 异常时可能无 JSON 输出
## 修复
- Agent 有 CPU/内存时优先用 RedisSSH 仅补充 IOSSH 失败仍显示 Agent 指标
- SSH 探针脚本增加 try/exceptJSON 解析支持行内 `{` 前缀噪声
- 前端 `parse_error` 显示为「探针解析失败」,并展示 `metrics.error` 详情
## 2026-06-11 补充:错误中文提示
- 新增 `server/utils/watch_probe_errors.py`,新探针失败写入中文 `error`
- 前端 `formatProbeError` / `probeStatusLabel` / `probeSourceLabel` 统一中文(含历史英文记录)
- 监测历史探针记录表:状态筛选与错误列中文化
## 涉及文件
- `server/background/watch_probe_runner.py`
- `server/utils/watch_metrics.py`
- `server/infrastructure/ssh/remote_probe.py`
- `frontend/src/utils/watchFormat.ts`
- `frontend/src/components/watch/WatchSlotCard.vue`
- `tests/test_watch_metrics.py`
## 验证
```bash
pytest tests/test_watch_metrics.py -q
cd frontend && npm run type-check
```
@@ -0,0 +1,55 @@
# 监测槽实时监测开关
**日期**: 2026-06-11
## 摘要
为每台已固定监测槽的服务器增加「实时监测」开关。关闭后停止 5 秒探针与 WebSocket 推送,槽位仍保留同一台服务器,8 小时倒计时暂停;重新开启时重置 8 小时窗口。
## 动机
用户希望随时关闭某台服务器的实时监测,而不必等到 8 小时到期或点击删除释放槽位。
## 涉及文件
- `server/domain/models/__init__.py``WatchPin.monitoring_enabled`
- `server/infrastructure/database/migrations.py` — 列迁移
- `server/infrastructure/database/watch_repo.py` — 暂停/恢复、过期逻辑
- `server/application/services/watch_service.py``set_slot_monitoring`、槽位 payload
- `server/api/watch.py``PATCH /api/watch/pins/{slot_index}/monitoring`
- `server/background/watch_probe_runner.py` — 仅探针已开启的 pin
- `frontend/src/api/index.ts``http.patch`
- `frontend/src/composables/useWatchPins.ts`
- `frontend/src/components/watch/WatchSlotCard.vue`
- `frontend/src/components/watch/WatchSlotRow.vue`
- `tests/test_watch_pins.py`
## 迁移 / 重启
- 需 API 重启以执行 schema migration`monitoring_enabled` 列)
- 无需手动 SQL
## 验证
```bash
pytest tests/test_watch_pins.py -q
cd frontend && npm run type-check
```
- 仪表盘监测槽:开关关闭 → 卡片灰显、无实时指标;重新开启 → 恢复探针并重置倒计时
- 时长切换:默认 8h;点击 24h → 重置为 24 小时窗口;暂停时切换时长保留偏好,开启后生效
## 2026-06-11 补充:暂停永久占槽 + 恢复 Agent 60s
- **暂停/时长到期**:不再删除 pin;`monitoring_enabled=false`,槽位永久保留直至手动移除或重新开启
- **8h/24h 到期**:自动暂停(等同关开关),结束 session(`expired`),保留槽位
- **恢复 Agent 模式**:当无任何槽位对该机开启 5s 监测时,清理 `watch:live/proc/last` Redis;子机 Agent 始终 60s 智能心跳(CPU 波动 >10% 或超阈值才上报),无需改 Agent 配置
- **重新开启**:若 session 已结束则自动开新 session
- **涉及**`server/utils/watch_state.py``watch_repo.expire_due_pins``watch_service``watch_probe_runner`、前端 countdown 到期 refresh
## 2026-06-11 补充:8h / 24h 时长选项
- `watch_pins.ttl_hours` 列(默认 8,仅允许 8 或 24)
- `PATCH /api/watch/pins/{slot_index}/ttl` — 切换时长;监测开启时同步重置 `expires_at`
- `POST /api/watch/pins` 可选 `ttl_hours: 8 | 24`(默认 8
- 槽位卡片「8h / 24h」切换按钮,须显式点击才改为 24h
@@ -0,0 +1,35 @@
# Changelog — 监测槽 TTL 到期竞态修复
**日期**2026-06-12
## 摘要
修复 TTL 到期与后台 `expire_due_pins`(5s 探针循环)之间的竞态窗口:槽位短暂显示为空、无法添加/移除、可能触发唯一约束冲突。
## 动机
`list_active_pins_for_admin``monitoring_enabled=true``expires_at<=now` 时排除 pin,但 DB 行仍在且 `monitoring_enabled` 仍为 true,直到探针循环执行 `expire_due_pins`。此窗口内 UI 显示空槽,但 `create_pin` 会因 `uq_watch_pins_admin_slot` 失败。
## 变更
- `WatchService._finalize_due_pins()`:在读/写路径同步到期 pin(暂停 + end session + 按需清 Redis
-`list_slots``pin_server``replace_slot``remove_slot``set_slot_monitoring``set_slot_ttl` 入口调用
- 新增单测:`test_watch_pin_ttl_expiry_list_syncs_before_display``test_watch_pin_ttl_expiry_pin_server_no_duplicate`
- **P2**`WatchSlotRow.onRemove` 失败时 snackbar 提示
## 涉及文件
- `server/application/services/watch_service.py`
- `frontend/src/components/watch/WatchSlotRow.vue`
- `tests/test_watch_pins.py`
## 迁移 / 重启
- 无需迁移
- 需重启 API
## 验证
```bash
.venv/bin/pytest tests/test_watch_pins.py -q
```
@@ -0,0 +1,40 @@
# Bug 巡查 — 监测槽增强(2026-06-12
**范围**:工作区未提交的 watch 开关 / TTL / 探针修复
**方法**:代码走读 + 单测 + 竞态分析
## 发现与处理
| # | 严重度 | 问题 | 状态 |
|---|--------|------|------|
| 1 | **P0** | TTL 到期后 ~5s 内槽位显示「空」,但 DB ghost pin 仍在;添加服务器 → 唯一约束冲突;点 X 移除 → 404「槽位为空」 | **已修** — API 读/写路径同步 `_finalize_due_pins` |
| 2 | P2 | `onRemove` 无 try/catch,API 失败时未提示 | 未修(低概率,可跟进) |
| 3 | P3 | 暂停态仍可通过 `get_processes` 拉取进程(数据可能过期) | 设计可接受 |
| 4 | — | Gate 7 被并行 Agent 提示改动阻塞 | 非 watch bug |
## P0 根因
```
expires_at 已过
→ list 查询 or_(paused, expires_at>now) 排除该行
→ UI empty=true
→ DB 仍有 monitoring_enabled=true 的行
→ create_pin / pick 槽位 → IntegrityError
```
探针循环每 5s 才跑 `expire_due_pins`,此前无 API 侧兜底。
## 修复
`WatchService._finalize_due_pins()``list_slots` 及所有槽位变更 API 入口调用,与探针循环逻辑一致:暂停 pin、结束 session、无活跃监测时清 Redis。
## 测试
```text
pytest tests/test_watch_pins.py → 7 passed
pytest tests/test_watch_*.py → 21 passed
```
## 未部署
生产仍为旧版本;修复需部署后浏览器终验 TTL 归零场景。
@@ -0,0 +1,100 @@
# 巡检 — 监测槽开关 / 8h·24h / 暂停占槽 / 探针修复
**日期**2026-06-12
**范围**:工作区未提交改动(监测槽增强 + parse_error 修复 + 中文错误)
**生产**https://api.synaglobal.vip**尚未部署本批改动**
## 进度条
| 步骤 | 状态 |
|------|------|
| 实现 | ✅ 工作区完成 |
| 本地验证 | ✅ L2b 26/26 + ruff |
| 单测 | ✅ watch 23/23 |
| 类型检查 | ✅ vue-tsc |
| 门控 7/7 | ⚠️ 6/7Gate 7 被 **Agent 提示** 并行改动阻塞,非 watch 本身) |
| 部署 | ❌ 未执行 |
| 生产健康 | ✅ `/health``ok` |
| 浏览器终验 | ❌ 待部署后 |
## 本地验证
### 单测
```text
pytest tests/test_watch_*.py tests/test_watch_probe_errors.py
→ 23 passed
```
### L2b API E2Elocal_verify.sh
```text
26/26 passed
ruff F: All checks passed
═══ All local checks passed ═══
```
### 前端
```text
npm run type-check → PASS
```
### 门控 pre_deploy_check.sh
| Gate | 结果 |
|------|------|
| 1 Changelog | PASS2026-06-12-server-agent-action-hint.md |
| 2 Audit | PASS2026-06-12-files-watch-ui-fixes.md |
| 3 Test | PASS |
| 4 Lint | PASS |
| 5 Import | PASS |
| 6 Security | PASS |
| 7 Review | **BLOCK** — 以下文件有 diff 但未列入当日 audit |
Gate 7 阻塞文件(**Agent 版本提示**,与监测槽无关):
- `frontend/src/components/servers/ServerAgentHint.vue`
- `frontend/src/components/servers/ServerInlineDetail.vue`
- `frontend/src/composables/servers/useServerFormDialog.ts`
- `frontend/src/types/api.ts`
- `server/api/servers.py`
- `server/utils/agent_version.py`
- `tests/test_agent_version.py`
监测槽改动已有独立审计:`docs/audit/2026-06-11-watch-slot-monitoring-enhancements.md`
## 生产探针
| 检查 | 结果 |
|------|------|
| `GET /health` | 200 `ok` |
| `GET /app/` | 200 |
| `GET /api/watch/pins`(无 JWT | 401(预期) |
| `PATCH /api/watch/pins/0/monitoring`(无 JWT) | 401(端点存在性需部署后带 JWT 终验) |
**说明**:生产当前运行的是已部署版本;监测槽开关 / TTL PATCH / 中文错误 / parse_error 修复均在工作区,**线上行为未变**。
## 功能验收清单(部署后浏览器终验)
1. **开关暂停**:关实时监测 → 5s 探针停止,槽位仍占用,UI 显示暂停态
2. **重新开启**:开开关 → 若 session 已结束则自动开新 session,探针恢复
3. **8h / 24h**:默认 8h;点 24h 后 `expires_at` 延长;replace 槽位保留原 TTL
4. **TTL 到期**:自动暂停(不删 pin),Agent 恢复 60s 智能心跳
5. **parse_error**Agent 有 CPU/内存时 SSH 失败不应整轮 parse_error
6. **中文错误**:探针失败状态与错误列显示中文
## 部署前待办
1. **Gate 7**:为 Agent 提示改动补 audit,或与监测槽分两次提交/部署
2. **迁移**`watch_pins.monitoring_enabled` + `ttl_hours`migrations.py 已含)
3. **部署**:用户批准后 `bash deploy/deploy-production.sh`
4. **前端构建**Docker 内 vite build 或 `deploy/deploy-frontend.sh`
## 本次巡检修复
- 删除 `server/infrastructure/ssh/remote_probe.py` 未使用的 `import logging`local_verify ruff F401
## 结论
**代码质量就绪**(单测 + L2b + lint + 类型检查均通过)。**部署被 Gate 7 阻塞**(并行 Agent 提示改动缺 audit)。监测槽功能本身审计完备,建议与 Agent 提示改动解耦提交,或补全 audit 后一并部署。
+3
View File
@@ -296,6 +296,9 @@ export const http = {
put: <T = any>(path: string, body?: unknown) =>
api<T>(path, { method: 'PUT', body: body ? JSON.stringify(body) : undefined }),
patch: <T = any>(path: string, body?: unknown) =>
api<T>(path, { method: 'PATCH', body: body ? JSON.stringify(body) : undefined }),
delete: <T = any>(path: string) =>
api<T>(path, { method: 'DELETE' }),
@@ -1,7 +1,7 @@
<script setup lang="ts">
import { ref, watch } from 'vue'
import { http } from '@/api'
import { formatBytesPerSec, probeStatusColor } from '@/utils/watchFormat'
import { formatBytesPerSec, probeStatusColor, probeStatusLabel, formatProbeError, probeSourceLabel, PROBE_STATUS_FILTER_ITEMS } from '@/utils/watchFormat'
export interface ProbeRecordRow {
id: number
@@ -75,12 +75,7 @@ watch(
<div class="d-flex align-center ga-2 mb-3">
<v-select
v-model="statusFilter"
:items="[
{ title: '全部状态', value: null },
{ title: 'ok', value: 'ok' },
{ title: 'ssh_timeout', value: 'ssh_timeout' },
{ title: 'no_cred', value: 'no_cred' },
]"
:items="PROBE_STATUS_FILTER_ITEMS"
item-title="title"
item-value="value"
label="探针状态"
@@ -105,14 +100,19 @@ watch(
>
<template #item.probe_status="{ item }">
<v-chip size="x-small" :color="probeStatusColor(item.probe_status)" variant="tonal" label>
{{ item.probe_status }}
{{ probeStatusLabel(item.probe_status) }}
</v-chip>
</template>
<template #item.source="{ item }">{{ probeSourceLabel(item.source) }}</template>
<template #item.net_up_bps="{ item }">{{ formatBytesPerSec(item.net_up_bps) }}</template>
<template #item.net_down_bps="{ item }">{{ formatBytesPerSec(item.net_down_bps) }}</template>
<template #item.duration_ms="{ item }">{{ item.duration_ms ?? '—' }} ms</template>
<template #item.error="{ item }">
<span class="text-caption text-truncate d-inline-block" style="max-width: 200px">{{ item.error || '—' }}</span>
<span
class="text-caption text-truncate d-inline-block"
style="max-width: 200px"
:title="formatProbeError(item.error, item.probe_status)"
>{{ formatProbeError(item.error, item.probe_status) }}</span>
</template>
</v-data-table>
<div class="d-flex justify-center align-center ga-2 mt-3">
@@ -1,7 +1,7 @@
<script setup lang="ts">
import { ref } from 'vue'
import { ref, computed } from 'vue'
import type { WatchSlot } from '@/composables/useWatchPins'
import { formatBytesPerSec, formatRemaining, probeStatusColor } from '@/utils/watchFormat'
import { formatBytesPerSec, formatRemaining, probeStatusColor, probeStatusLabel, formatProbeError } from '@/utils/watchFormat'
import WatchSparkline from '@/components/watch/WatchSparkline.vue'
const props = defineProps<{
@@ -13,8 +13,13 @@ const emit = defineEmits<{
processes: [slot: WatchSlot]
history: [slot: WatchSlot]
pick: [slotIndex: number]
monitoring: [slotIndex: number, enabled: boolean]
ttl: [slotIndex: number, hours: 8 | 24]
}>()
const monitoringOn = computed(() => props.slot.monitoring_enabled !== false)
const slotTtl = computed(() => (props.slot.ttl_hours === 24 ? 24 : 8) as 8 | 24)
const chartMetric = ref<'cpu_pct' | 'mem_pct' | 'disk_pct'>('cpu_pct')
function pct(v: number | null | undefined) {
@@ -29,7 +34,10 @@ function pct(v: number | null | undefined) {
border
rounded="lg"
class="watch-slot-card pa-3 watch-slot-card--filled"
:class="{ 'watch-slot-card--error': slot.metrics?.probe_status && slot.metrics.probe_status !== 'ok' }"
:class="{
'watch-slot-card--error': monitoringOn && slot.metrics?.probe_status && slot.metrics.probe_status !== 'ok',
'watch-slot-card--paused': !monitoringOn,
}"
role="button"
tabindex="0"
@click="emit('history', slot)"
@@ -42,9 +50,38 @@ function pct(v: number | null | undefined) {
<v-chip size="x-small" variant="tonal" color="info" label>
{{ slot.slot_index + 1 }}
</v-chip>
<v-switch
:model-value="monitoringOn"
density="compact"
hide-details
color="primary"
class="watch-slot-switch ml-1"
@click.stop
@update:model-value="emit('monitoring', slot.slot_index, $event ?? false)"
/>
<v-btn icon="mdi-close" size="x-small" variant="text" @click="emit('remove', slot.slot_index)" />
</div>
<div class="d-flex align-center mb-2 ga-1" @click.stop>
<span class="text-caption text-medium-emphasis mr-1">时长</span>
<v-btn-toggle
:model-value="slotTtl"
density="compact"
variant="outlined"
divided
mandatory
@update:model-value="emit('ttl', slot.slot_index, ($event ?? 8) as 8 | 24)"
>
<v-btn :value="8" size="x-small">8h</v-btn>
<v-btn :value="24" size="x-small">24h</v-btn>
</v-btn-toggle>
</div>
<div v-if="!monitoringOn" class="text-caption text-medium-emphasis mb-2" @click.stop>
实时监测已关闭 · 槽位保留 · Agent 恢复 60s 智能上报
</div>
<template v-if="monitoringOn">
<div @click.stop>
<v-chip
v-if="slot.metrics?.probe_status"
@@ -54,7 +91,7 @@ function pct(v: number | null | undefined) {
class="mb-2"
label
>
{{ slot.metrics.probe_status === 'ok' ? '正常' : slot.metrics.probe_status }}
{{ probeStatusLabel(slot.metrics.probe_status) }}
</v-chip>
<div class="d-flex justify-space-between text-caption mb-1">
@@ -80,18 +117,32 @@ function pct(v: number | null | undefined) {
<WatchSparkline :points="slot.sparkline || []" :metric="chartMetric" />
<div v-if="slot.metrics?.error" class="text-caption text-error mt-1 text-truncate">
{{ slot.metrics.error }}
<div
v-if="slot.metrics?.error"
class="text-caption text-error mt-1 text-truncate"
:title="formatProbeError(slot.metrics.error, slot.metrics.probe_status)"
>
{{ formatProbeError(slot.metrics.error, slot.metrics.probe_status) }}
</div>
<div class="d-flex align-center justify-space-between mt-2">
<span class="text-caption text-medium-emphasis">剩余 {{ formatRemaining(slot.remaining_sec) }}</span>
<span class="text-caption text-medium-emphasis">
剩余 {{ formatRemaining(slot.remaining_sec) }} · {{ slotTtl }}h
</span>
<div class="d-flex ga-1">
<v-btn size="x-small" variant="text" @click.stop="emit('processes', slot)">进程</v-btn>
<v-btn size="x-small" variant="text" @click.stop="emit('history', slot)">历史</v-btn>
</div>
</div>
</div>
</template>
<div v-else class="d-flex align-center justify-space-between mt-2" @click.stop>
<span class="text-caption text-medium-emphasis">已暂停 · {{ slotTtl }}h</span>
<div class="d-flex ga-1">
<v-btn size="x-small" variant="text" @click.stop="emit('history', slot)">历史</v-btn>
</div>
</div>
</v-card>
<v-card
@@ -135,4 +186,12 @@ function pct(v: number | null | undefined) {
.watch-slot-card--error {
border-color: rgb(var(--v-theme-error)) !important;
}
.watch-slot-card--paused {
opacity: 0.72;
background: rgba(var(--v-theme-surface-variant), 0.15);
}
.watch-slot-switch {
flex: 0 0 auto;
transform: scale(0.85);
}
</style>
+31 -3
View File
@@ -6,10 +6,11 @@ import WatchProcessDrawer from '@/components/watch/WatchProcessDrawer.vue'
import WatchSlotPickDialog from '@/components/watch/WatchSlotPickDialog.vue'
import { useWatchPins, type WatchSlot } from '@/composables/useWatchPins'
import { useSnackbar } from '@/composables/useSnackbar'
import { formatApiError } from '@/utils/apiError'
const router = useRouter()
const snackbar = useSnackbar()
const { slots, loading, refreshPins, removeSlot, pinServer } = useWatchPins()
const { slots, loading, refreshPins, removeSlot, pinServer, setSlotMonitoring, setSlotTtl } = useWatchPins()
const processDrawer = ref(false)
const processSlot = ref<WatchSlot | null>(null)
@@ -22,7 +23,32 @@ onMounted(() => {
})
async function onRemove(slotIndex: number) {
await removeSlot(slotIndex)
try {
await removeSlot(slotIndex)
snackbar(`已移除监测 · 槽 ${slotIndex + 1}`, 'success')
} catch (e: unknown) {
snackbar(formatApiError(e, '移除监测失败'), 'error')
}
}
async function onMonitoring(slotIndex: number, enabled: boolean) {
try {
await setSlotMonitoring(slotIndex, enabled)
snackbar(enabled ? `已开启实时监测 · 槽 ${slotIndex + 1}` : `已暂停实时监测 · 槽 ${slotIndex + 1}`, 'success')
} catch (e: unknown) {
snackbar(formatApiError(e, '更新监测开关失败'), 'error')
}
}
async function onTtl(slotIndex: number, hours: 8 | 24) {
const slot = slots.value.find((s) => s.slot_index === slotIndex)
if (!slot || slot.empty || slot.ttl_hours === hours) return
try {
await setSlotTtl(slotIndex, hours)
snackbar(`监测时长已设为 ${hours} 小时 · 槽 ${slotIndex + 1}`, 'success')
} catch (e: unknown) {
snackbar(formatApiError(e, '更新监测时长失败'), 'error')
}
}
function openProcesses(slot: WatchSlot) {
@@ -68,7 +94,7 @@ async function onPickServer(serverId: number) {
<div class="watch-slot-row mb-4">
<div class="d-flex align-center mb-2">
<span class="text-subtitle-2">实时监测</span>
<v-chip size="x-small" variant="tonal" class="ml-2" label>最多 4 · 8h</v-chip>
<v-chip size="x-small" variant="tonal" class="ml-2" label>最多 4 · 默认 8h · 暂停不占 TTL</v-chip>
<v-spacer />
<v-btn size="small" variant="text" prepend-icon="mdi-history" :to="{ name: 'WatchMetrics' }">
监测历史
@@ -83,6 +109,8 @@ async function onPickServer(serverId: number) {
@processes="openProcesses"
@history="openHistory"
@pick="onPickSlot"
@monitoring="onMonitoring"
@ttl="onTtl"
/>
</v-col>
</v-row>
+43 -17
View File
@@ -48,6 +48,8 @@ export interface WatchSlot {
pinned_at?: string
expires_at?: string
remaining_sec?: number
monitoring_enabled?: boolean
ttl_hours?: 8 | 24
metrics?: WatchMetrics | null
sparkline?: WatchSparkPoint[]
processes?: WatchProcess[]
@@ -73,34 +75,38 @@ function applySlots(data: { slots?: WatchSlot[] }) {
}
function tickCountdown() {
let shouldRefresh = false
for (const slot of slots.value) {
if (!slot.empty && slot.remaining_sec != null && slot.remaining_sec > 0) {
if (slot.empty || slot.monitoring_enabled === false) continue
if (slot.remaining_sec != null && slot.remaining_sec > 0) {
slot.remaining_sec -= 1
if (slot.remaining_sec <= 0) {
slot.empty = true
shouldRefresh = true
}
}
}
if (shouldRefresh) {
void refreshPins()
}
}
function mergeMetrics(serverId: number, metrics: WatchMetrics) {
for (const slot of slots.value) {
if (!slot.empty && slot.server_id === serverId) {
slot.metrics = { ...(slot.metrics || {}), ...metrics }
if (metrics.processes?.length) {
slot.processes = metrics.processes
}
const point: WatchSparkPoint = {
ts: metrics.ts,
cpu_pct: metrics.cpu_pct,
mem_pct: metrics.mem_pct,
disk_pct: metrics.disk_pct,
net_down_bps: metrics.net_down_bps,
}
if (!slot.sparkline) slot.sparkline = []
slot.sparkline.push(point)
if (slot.sparkline.length > 180) slot.sparkline.shift()
if (slot.empty || slot.server_id !== serverId || slot.monitoring_enabled === false) continue
slot.metrics = { ...(slot.metrics || {}), ...metrics }
if (metrics.processes?.length) {
slot.processes = metrics.processes
}
const point: WatchSparkPoint = {
ts: metrics.ts,
cpu_pct: metrics.cpu_pct,
mem_pct: metrics.mem_pct,
disk_pct: metrics.disk_pct,
net_down_bps: metrics.net_down_bps,
}
if (!slot.sparkline) slot.sparkline = []
slot.sparkline.push(point)
if (slot.sparkline.length > 180) slot.sparkline.shift()
}
}
@@ -182,6 +188,24 @@ async function removeSlot(slotIndex: number) {
applySlots(data)
}
async function setSlotMonitoring(slotIndex: number, monitoringEnabled: boolean) {
const data = await http.patch<{ slots: WatchSlot[] }>(
`/watch/pins/${slotIndex}/monitoring`,
{ monitoring_enabled: monitoringEnabled },
)
applySlots(data)
}
export type WatchTtlHours = 8 | 24
async function setSlotTtl(slotIndex: number, ttlHours: WatchTtlHours) {
const data = await http.patch<{ slots: WatchSlot[] }>(
`/watch/pins/${slotIndex}/ttl`,
{ ttl_hours: ttlHours },
)
applySlots(data)
}
export function useWatchPins() {
consumerCount += 1
if (consumerCount === 1) {
@@ -223,5 +247,7 @@ export function useWatchPins() {
refreshPins,
pinServer,
removeSlot,
setSlotMonitoring,
setSlotTtl,
}
}
+89
View File
@@ -15,8 +15,97 @@ export function formatRemaining(sec: number | null | undefined): string {
return `${m}m ${sec % 60}s`
}
const PROBE_STATUS_LABELS: Record<string, string> = {
ok: '正常',
parse_error: '探针解析失败',
ssh_timeout: 'SSH 超时',
ssh_auth: 'SSH 认证失败',
no_cred: '无凭据',
redis_stale: 'Agent 心跳过期',
offline: '离线',
}
const PROBE_ERROR_EXACT: Record<string, string> = {
parse_error: '探针返回数据无法解析',
'psutil missing': '子机未安装 psutil(请执行 pip install psutil',
ssh_probe_failed: 'SSH 探针脚本执行失败',
'SSH watch probe failed': 'SSH 监测探针失败',
'SSH probe failed': 'SSH 探针失败',
timeout: '连接超时',
'探针失败': '探针失败',
'无 Agent 且 SSH 探针失败': '无 Agent 且 SSH 探针失败',
'服务器不存在': '服务器不存在',
}
/** Probe status code → Chinese label. */
export function probeStatusLabel(status: string | null | undefined): string {
if (!status || status === 'ok') return PROBE_STATUS_LABELS.ok
if (PROBE_STATUS_LABELS[status]) return PROBE_STATUS_LABELS[status]
if (status.startsWith('parse_error')) return PROBE_STATUS_LABELS.parse_error
return status
}
/** Probe source code → Chinese label. */
export function probeSourceLabel(source: string | null | undefined): string {
if (!source) return '—'
const map: Record<string, string> = {
redis: 'Agent',
ssh: 'SSH',
mixed: 'Agent+SSH',
}
return map[source] ?? source
}
/** Error detail → Chinese (covers legacy English rows in DB). */
export function formatProbeError(
error: string | null | undefined,
status?: string | null,
): string {
if (!error?.trim()) {
if (status && status !== 'ok') return probeStatusLabel(status)
return '—'
}
const text = error.trim()
if (PROBE_ERROR_EXACT[text]) return PROBE_ERROR_EXACT[text]
const lower = text.toLowerCase()
if (lower.startsWith('parse_error')) {
const detail = text.includes(':') ? text.split(':').slice(1).join(':').trim() : ''
if (detail && PROBE_ERROR_EXACT[detail]) {
return `探针返回数据无法解析:${PROBE_ERROR_EXACT[detail]}`
}
if (detail) return `探针返回数据无法解析:${detail}`
return PROBE_ERROR_EXACT.parse_error
}
if (lower.includes('timed out') || lower.endsWith('timeout')) {
return `SSH 连接超时:${text}`
}
if (lower.includes('permission denied') || lower.includes('authentication failed')) {
return `SSH 认证失败:${text}`
}
if (lower.includes('psutil missing')) {
return PROBE_ERROR_EXACT['psutil missing']
}
if (lower.includes('ssh_probe_failed') || lower.includes('ssh watch probe failed')) {
return PROBE_ERROR_EXACT.ssh_probe_failed
}
if (/[\u4e00-\u9fff]/.test(text)) return text
return `探针异常:${text}`
}
export function probeStatusColor(status: string | null | undefined): string {
if (!status || status === 'ok') return 'success'
if (status.includes('timeout') || status.includes('auth')) return 'warning'
return 'error'
}
/** Status filter options for probe records table. */
export const PROBE_STATUS_FILTER_ITEMS = [
{ title: '全部状态', value: null as string | null },
{ title: '正常', value: 'ok' },
{ title: 'SSH 超时', value: 'ssh_timeout' },
{ title: '探针解析失败', value: 'parse_error' },
{ title: '无凭据', value: 'no_cred' },
{ title: '离线', value: 'offline' },
]
+56
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
from typing import Literal
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
@@ -9,6 +11,7 @@ from server.api.auth_jwt import get_current_admin
from server.api.dependencies import get_db
from server.application.services.watch_service import SlotsFullError, WatchService
from server.domain.models import Admin
from server.utils.watch_metrics import WATCH_PIN_TTL_HOURS
from sqlalchemy.ext.asyncio import AsyncSession
router = APIRouter(prefix="/api/watch", tags=["watch"])
@@ -18,12 +21,24 @@ class WatchPinCreate(BaseModel):
server_id: int = Field(..., gt=0)
slot_index: int | None = Field(None, ge=0, le=3)
replace_slot: int | None = Field(None, ge=0, le=3, description="监测已满时指定替换槽位")
ttl_hours: Literal[8, 24] = Field(
WATCH_PIN_TTL_HOURS,
description="监测窗口时长,默认 8 小时;24 小时需显式指定",
)
class WatchPinReplace(BaseModel):
server_id: int = Field(..., gt=0)
class WatchPinMonitoringUpdate(BaseModel):
monitoring_enabled: bool = Field(..., description="是否开启此槽位实时监测")
class WatchPinTtlUpdate(BaseModel):
ttl_hours: Literal[8, 24] = Field(..., description="监测窗口时长:8 或 24 小时")
def _client_ip(request: Request) -> str:
return request.client.host if request.client else ""
@@ -51,6 +66,7 @@ async def create_watch_pin(
server_id=payload.server_id,
slot_index=payload.slot_index,
replace_slot=payload.replace_slot,
ttl_hours=payload.ttl_hours,
ip_address=_client_ip(request),
)
except SlotsFullError as e:
@@ -97,6 +113,46 @@ async def delete_watch_pin(
raise HTTPException(status_code=404, detail=str(e)) from e
@router.patch("/pins/{slot_index}/monitoring")
async def update_watch_pin_monitoring(
slot_index: int,
payload: WatchPinMonitoringUpdate,
request: Request,
admin: Admin = Depends(get_current_admin),
db: AsyncSession = Depends(get_db),
):
svc = WatchService(db)
try:
return await svc.set_slot_monitoring(
admin=admin,
slot_index=slot_index,
monitoring_enabled=payload.monitoring_enabled,
ip_address=_client_ip(request),
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.patch("/pins/{slot_index}/ttl")
async def update_watch_pin_ttl(
slot_index: int,
payload: WatchPinTtlUpdate,
request: Request,
admin: Admin = Depends(get_current_admin),
db: AsyncSession = Depends(get_db),
):
svc = WatchService(db)
try:
return await svc.set_slot_ttl(
admin=admin,
slot_index=slot_index,
ttl_hours=payload.ttl_hours,
ip_address=_client_ip(request),
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
@router.get("/metrics")
async def get_watch_metrics(
server_ids: str = Query(..., description="逗号分隔 server_id"),
+123 -18
View File
@@ -13,7 +13,13 @@ from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
from server.infrastructure.database.server_repo import ServerRepositoryImpl
from server.infrastructure.database.watch_repo import WatchRepositoryImpl
from server.infrastructure.redis.client import get_redis
from server.utils.watch_metrics import WATCH_MAX_SLOTS, WATCH_PIN_TTL_HOURS, server_can_be_watched
from server.utils.watch_state import clear_watch_redis_snapshot
from server.utils.watch_metrics import (
WATCH_MAX_SLOTS,
WATCH_PIN_TTL_HOURS,
normalize_watch_ttl_hours,
server_can_be_watched,
)
REDIS_WATCH_LIVE_PREFIX = "watch:live:"
REDIS_WATCH_PROC_PREFIX = "watch:proc:"
@@ -67,6 +73,23 @@ async def _load_processes(server_id: int) -> list[dict[str, Any]] | None:
return None
async def _restore_idle_agent_watch(watch_repo: WatchRepositoryImpl, server_id: int) -> None:
"""Stop 5s watch snapshots when no slot actively monitors this server (Agent stays 60s)."""
if await watch_repo.count_monitoring_pins_for_server(server_id) == 0:
await clear_watch_redis_snapshot(server_id)
async def _finalize_due_pins(watch_repo: WatchRepositoryImpl, db: AsyncSession) -> None:
"""Pause TTL-expired pins on read/write paths (probe loop also runs this every 5s)."""
expired_servers = await watch_repo.expire_due_pins()
if not expired_servers:
return
for server_id in expired_servers:
if await watch_repo.count_monitoring_pins_for_server(server_id) == 0:
await clear_watch_redis_snapshot(server_id)
await db.commit()
async def _slot_payload(
*,
slot_index: int,
@@ -74,11 +97,17 @@ async def _slot_payload(
) -> dict[str, Any]:
if not pin_row:
return {"slot_index": slot_index, "empty": True}
metrics = await _load_live_metrics(pin_row["server_id"])
sparkline = await _load_sparkline(pin_row["server_id"]) if pin_row else []
processes = await _load_processes(pin_row["server_id"])
if metrics and processes:
metrics = {**metrics, "processes": processes}
monitoring_enabled = pin_row.get("monitoring_enabled", True)
metrics = None
sparkline: list[dict[str, Any]] = []
processes = None
if monitoring_enabled:
metrics = await _load_live_metrics(pin_row["server_id"])
sparkline = await _load_sparkline(pin_row["server_id"])
processes = await _load_processes(pin_row["server_id"])
if metrics and processes:
metrics = {**metrics, "processes": processes}
remaining = _remaining_sec(pin_row["expires_at"]) if monitoring_enabled else None
return {
"slot_index": slot_index,
"empty": False,
@@ -88,7 +117,9 @@ async def _slot_payload(
"session_id": pin_row["session_id"],
"pinned_at": pin_row["pinned_at"].isoformat() if pin_row.get("pinned_at") else None,
"expires_at": pin_row["expires_at"].isoformat() if pin_row.get("expires_at") else None,
"remaining_sec": _remaining_sec(pin_row["expires_at"]),
"remaining_sec": remaining,
"monitoring_enabled": monitoring_enabled,
"ttl_hours": int(pin_row.get("ttl_hours") or WATCH_PIN_TTL_HOURS),
"metrics": metrics,
"sparkline": sparkline,
"processes": processes,
@@ -103,6 +134,7 @@ class WatchService:
self.audit_repo = AuditLogRepositoryImpl(db)
async def list_slots(self, admin_id: int) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
pins = await self.watch_repo.list_active_pins_for_admin(admin_id)
by_slot = {p["slot_index"]: p for p in pins}
slots = [
@@ -118,8 +150,11 @@ class WatchService:
server_id: int,
slot_index: int | None = None,
replace_slot: int | None = None,
ttl_hours: int = WATCH_PIN_TTL_HOURS,
ip_address: str = "",
) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
hours = normalize_watch_ttl_hours(ttl_hours)
server = await self.server_repo.get_by_id(server_id)
if not server:
raise ValueError("服务器不存在")
@@ -130,12 +165,12 @@ class WatchService:
existing = await self.watch_repo.get_pin_by_server(admin.id, server_id)
if existing:
await self.watch_repo.reset_pin_ttl(admin.id, server_id, WATCH_PIN_TTL_HOURS)
await self.watch_repo.reset_pin_ttl(admin.id, server_id, hours)
await self._audit(
admin=admin,
action="watch_pin_add",
server=server,
detail=f"重置监测 8h · 槽 {existing.slot_index + 1}",
detail=f"重置监测 {hours}h · 槽 {existing.slot_index + 1}",
ip_address=ip_address,
)
await self.db.commit()
@@ -146,13 +181,13 @@ class WatchService:
admin_id=admin.id,
slot_index=slot_index,
server_id=server_id,
ttl_hours=WATCH_PIN_TTL_HOURS,
ttl_hours=hours,
)
await self._audit(
admin=admin,
action="watch_pin_replace" if replace_slot is not None else "watch_pin_add",
server=server,
detail=f"加入监测 · 槽 {slot_index + 1}",
detail=f"加入监测 · 槽 {slot_index + 1} · {hours}h",
ip_address=ip_address,
)
await self.db.commit()
@@ -164,13 +199,13 @@ class WatchService:
admin_id=admin.id,
server_id=server_id,
slot_index=empty,
ttl_hours=WATCH_PIN_TTL_HOURS,
ttl_hours=hours,
)
await self._audit(
admin=admin,
action="watch_pin_add",
server=server,
detail=f"加入监测 · 槽 {empty + 1}",
detail=f"加入监测 · 槽 {empty + 1} · {hours}h",
ip_address=ip_address,
)
await self.db.commit()
@@ -181,13 +216,13 @@ class WatchService:
admin_id=admin.id,
slot_index=replace_slot,
server_id=server_id,
ttl_hours=WATCH_PIN_TTL_HOURS,
ttl_hours=hours,
)
await self._audit(
admin=admin,
action="watch_pin_replace",
server=server,
detail=f"替换监测 · 槽 {replace_slot + 1}",
detail=f"替换监测 · 槽 {replace_slot + 1} · {hours}h",
ip_address=ip_address,
)
await self.db.commit()
@@ -203,6 +238,7 @@ class WatchService:
server_id: int,
ip_address: str = "",
) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
if slot_index < 0 or slot_index >= WATCH_MAX_SLOTS:
raise ValueError("槽位无效")
server = await self.server_repo.get_by_id(server_id)
@@ -215,17 +251,19 @@ class WatchService:
if other and other.slot_index != slot_index:
await self.watch_repo.end_session(other.session_id, "replaced")
await self.watch_repo.delete_pin(other.id)
existing = await self.watch_repo.get_pin_by_slot(admin.id, slot_index)
slot_ttl = normalize_watch_ttl_hours(int(existing.ttl_hours or 8)) if existing else WATCH_PIN_TTL_HOURS
await self.watch_repo.replace_pin_server(
admin_id=admin.id,
slot_index=slot_index,
server_id=server_id,
ttl_hours=WATCH_PIN_TTL_HOURS,
ttl_hours=slot_ttl,
)
await self._audit(
admin=admin,
action="watch_pin_replace",
server=server,
detail=f"替换监测 · 槽 {slot_index + 1}",
detail=f"替换监测 · 槽 {slot_index + 1} · {slot_ttl}h",
ip_address=ip_address,
)
await self.db.commit()
@@ -238,12 +276,14 @@ class WatchService:
slot_index: int,
ip_address: str = "",
) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
if slot_index < 0 or slot_index >= WATCH_MAX_SLOTS:
raise ValueError("槽位无效")
pin = await self.watch_repo.remove_pin_by_slot(admin.id, slot_index)
if not pin:
raise ValueError("槽位为空")
server = await self.server_repo.get_by_id(pin.server_id)
server_id = pin.server_id
server = await self.server_repo.get_by_id(server_id)
await self._audit(
admin=admin,
action="watch_pin_remove",
@@ -252,6 +292,71 @@ class WatchService:
ip_address=ip_address,
)
await self.db.commit()
await _restore_idle_agent_watch(self.watch_repo, server_id)
return await self.list_slots(admin.id)
async def set_slot_monitoring(
self,
*,
admin: Admin,
slot_index: int,
monitoring_enabled: bool,
ip_address: str = "",
) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
if slot_index < 0 or slot_index >= WATCH_MAX_SLOTS:
raise ValueError("槽位无效")
pin = await self.watch_repo.set_monitoring_enabled(
admin.id,
slot_index,
monitoring_enabled,
)
if not pin:
raise ValueError("槽位为空")
server = await self.server_repo.get_by_id(pin.server_id)
hours = int(pin.ttl_hours or WATCH_PIN_TTL_HOURS)
action = "watch_pin_resume" if monitoring_enabled else "watch_pin_pause"
detail = (
f"开启实时监测 · 槽 {slot_index + 1} · 重置 {hours}h"
if monitoring_enabled
else f"暂停实时监测 · 槽 {slot_index + 1} · 恢复 Agent 60s"
)
await self._audit(
admin=admin,
action=action,
server=server,
detail=detail,
ip_address=ip_address,
)
await self.db.commit()
if not monitoring_enabled:
await _restore_idle_agent_watch(self.watch_repo, pin.server_id)
return await self.list_slots(admin.id)
async def set_slot_ttl(
self,
*,
admin: Admin,
slot_index: int,
ttl_hours: int,
ip_address: str = "",
) -> dict[str, Any]:
await _finalize_due_pins(self.watch_repo, self.db)
if slot_index < 0 or slot_index >= WATCH_MAX_SLOTS:
raise ValueError("槽位无效")
hours = normalize_watch_ttl_hours(ttl_hours)
pin = await self.watch_repo.set_pin_ttl_hours(admin.id, slot_index, hours)
if not pin:
raise ValueError("槽位为空")
server = await self.server_repo.get_by_id(pin.server_id)
await self._audit(
admin=admin,
action="watch_pin_ttl",
server=server,
detail=f"监测时长 · 槽 {slot_index + 1} · {hours}h",
ip_address=ip_address,
)
await self.db.commit()
return await self.list_slots(admin.id)
async def get_live_for_servers(self, server_ids: list[int]) -> dict[str, Any]:
+24 -7
View File
@@ -17,6 +17,8 @@ from server.infrastructure.database.watch_repo import WatchRepositoryImpl
from server.infrastructure.redis.client import get_redis
from server.infrastructure.ssh.remote_probe import ssh_watch_probe, ssh_watch_processes
from server.utils.watch_alerts import process_watch_probe_alerts
from server.utils.watch_probe_errors import watch_probe_error_zh
from server.utils.watch_state import clear_watch_redis_snapshot
from server.utils.watch_metrics import (
WATCH_PROBE_INTERVAL_SEC,
WATCH_PROBE_TIMEOUT_SEC,
@@ -29,6 +31,7 @@ from server.utils.watch_metrics import (
parse_redis_watch_payload,
parse_ssh_watch_payload,
redis_sample_has_io,
redis_sample_has_basic_metrics,
)
logger = logging.getLogger("nexus.watch_probe")
@@ -118,9 +121,17 @@ async def probe_server_metrics(
if heartbeat and _heartbeat_fresh(heartbeat):
redis_sample = parse_redis_watch_payload(heartbeat)
# M3: skip SSH when Agent heartbeat already carries full IO counters
if redis_sample_has_io(redis_sample):
sample = redis_sample # type: ignore[assignment]
if redis_sample is not None and redis_sample_has_io(redis_sample):
sample = redis_sample
elif redis_sample is not None and redis_sample_has_basic_metrics(redis_sample):
# Agent CPU/mem available — SSH only supplements IO; SSH failure must not block display
ssh_result = await ssh_watch_probe(server, timeout=WATCH_PROBE_TIMEOUT_SEC)
if ssh_result.get("ok"):
ssh_sample = parse_ssh_watch_payload(ssh_result["payload"])
ssh_sample.duration_ms = int(ssh_result.get("duration_ms") or 0)
sample = merge_redis_and_ssh(redis_sample, ssh_sample)
else:
sample = redis_sample
else:
ssh_result = await ssh_watch_probe(server, timeout=WATCH_PROBE_TIMEOUT_SEC)
if ssh_result.get("ok"):
@@ -139,13 +150,16 @@ async def probe_server_metrics(
return failure_sample(
status="no_cred",
source="ssh",
error=ssh_result.get("error") or "无 Agent 且 SSH 探针失败",
error=watch_probe_error_zh(
ssh_result.get("error") or "无 Agent 且 SSH 探针失败",
status="no_cred",
) or "无 Agent 且 SSH 探针失败",
duration_ms=int(ssh_result.get("duration_ms") or 0),
)
return failure_sample(
status=status,
source="ssh",
error=ssh_result.get("error") or "探针失败",
error=watch_probe_error_zh(ssh_result.get("error") or "探针失败", status=status) or "探针失败",
duration_ms=int(ssh_result.get("duration_ms") or 0),
)
@@ -161,8 +175,11 @@ async def _run_probe_cycle() -> None:
async with AsyncSessionLocal() as db:
watch_repo = WatchRepositoryImpl(db)
server_repo = ServerRepositoryImpl(db)
await watch_repo.expire_due_pins()
pins = await watch_repo.list_active_pins_all()
expired_servers = await watch_repo.expire_due_pins()
for server_id in expired_servers:
if await watch_repo.count_monitoring_pins_for_server(server_id) == 0:
await clear_watch_redis_snapshot(server_id)
pins = await watch_repo.list_active_pins_all(monitoring_only=True)
if not pins:
await db.commit()
return
+12
View File
@@ -183,6 +183,18 @@ class WatchPin(Base):
server_id = Column(Integer, ForeignKey("servers.id", ondelete="CASCADE"), nullable=False)
pinned_at = Column(DateTime, nullable=False, comment="UTC naive")
expires_at = Column(DateTime, nullable=False, comment="UTC naive")
monitoring_enabled = Column(
Boolean,
nullable=False,
default=True,
comment="是否开启实时探针(关闭时保留槽位、暂停倒计时)",
)
ttl_hours = Column(
Integer,
nullable=False,
default=8,
comment="监测窗口时长(小时),仅允许 8 或 24",
)
__table_args__ = (
Index("uq_watch_pins_admin_slot", "admin_id", "slot_index", unique=True),
@@ -239,6 +239,8 @@ async def run_schema_migrations():
`server_id` INT NOT NULL,
`pinned_at` DATETIME NOT NULL,
`expires_at` DATETIME NOT NULL,
`monitoring_enabled` TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否开启实时探针',
`ttl_hours` INT NOT NULL DEFAULT 8 COMMENT '监测窗口时长(小时) 8|24',
UNIQUE KEY `uq_watch_pins_admin_slot` (`admin_id`, `slot_index`),
UNIQUE KEY `uq_watch_pins_admin_server` (`admin_id`, `server_id`),
INDEX `idx_watch_pins_expires` (`expires_at`),
@@ -281,6 +283,8 @@ async def run_schema_migrations():
CONSTRAINT `fk_watch_probe_admin` FOREIGN KEY (`admin_id`) REFERENCES `admins` (`id`) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"ALTER TABLE `watch_probe_records` ADD COLUMN `processes_json` JSON NULL COMMENT 'Top5 processes snapshot'",
"ALTER TABLE `watch_pins` ADD COLUMN `monitoring_enabled` TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否开启实时探针'",
"ALTER TABLE `watch_pins` ADD COLUMN `ttl_hours` INT NOT NULL DEFAULT 8 COMMENT '监测窗口时长(小时) 8|24'",
]
async with AsyncSessionLocal() as session:
for sql in migrations:
+122 -29
View File
@@ -5,16 +5,24 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any
from sqlalchemy import delete, func, select
from sqlalchemy import delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from server.domain.models import Server, WatchPin, WatchPinSession, WatchProbeRecord
from server.utils.watch_metrics import normalize_watch_ttl_hours
def _utc_naive_now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
def _pin_still_occupied(pin: WatchPin, now: datetime) -> bool:
"""Pin occupies a slot until manual remove or TTL expiry while monitoring is on."""
if not pin.monitoring_enabled:
return True
return pin.expires_at > now
class WatchRepositoryImpl:
def __init__(self, session: AsyncSession):
self.session = session
@@ -25,12 +33,17 @@ class WatchRepositoryImpl:
select(WatchPin, WatchPinSession, Server.name)
.join(WatchPinSession, WatchPin.session_id == WatchPinSession.id)
.join(Server, WatchPin.server_id == Server.id)
.where(WatchPin.admin_id == admin_id, WatchPin.expires_at > now)
.where(
WatchPin.admin_id == admin_id,
or_(WatchPin.monitoring_enabled.is_(False), WatchPin.expires_at > now),
)
.order_by(WatchPin.slot_index)
)
rows = (await self.session.execute(q)).all()
out: list[dict[str, Any]] = []
for pin, sess, server_name in rows:
if not _pin_still_occupied(pin, now):
continue
out.append(
{
"pin_id": pin.id,
@@ -40,46 +53,55 @@ class WatchRepositoryImpl:
"session_id": sess.id,
"pinned_at": pin.pinned_at,
"expires_at": pin.expires_at,
"monitoring_enabled": bool(pin.monitoring_enabled),
"ttl_hours": int(pin.ttl_hours or 8),
}
)
return out
async def list_active_pins_all(self) -> list[dict[str, Any]]:
async def list_active_pins_all(self, *, monitoring_only: bool = False) -> list[dict[str, Any]]:
now = _utc_naive_now()
q = (
select(WatchPin, WatchPinSession)
.join(WatchPinSession, WatchPin.session_id == WatchPinSession.id)
.where(WatchPin.expires_at > now)
)
rows = (await self.session.execute(q)).all()
return [
{
"pin_id": pin.id,
"admin_id": pin.admin_id,
"slot_index": pin.slot_index,
"server_id": pin.server_id,
"session_id": sess.id,
}
for pin, sess in rows
]
out: list[dict[str, Any]] = []
for pin, sess in rows:
if not _pin_still_occupied(pin, now):
continue
if monitoring_only and not pin.monitoring_enabled:
continue
out.append(
{
"pin_id": pin.id,
"admin_id": pin.admin_id,
"slot_index": pin.slot_index,
"server_id": pin.server_id,
"session_id": sess.id,
}
)
return out
async def get_pin_by_slot(self, admin_id: int, slot_index: int) -> WatchPin | None:
now = _utc_naive_now()
q = select(WatchPin).where(
WatchPin.admin_id == admin_id,
WatchPin.slot_index == slot_index,
WatchPin.expires_at > now,
)
return (await self.session.execute(q)).scalar_one_or_none()
pin = (await self.session.execute(q)).scalar_one_or_none()
if pin and _pin_still_occupied(pin, _utc_naive_now()):
return pin
return None
async def get_pin_by_server(self, admin_id: int, server_id: int) -> WatchPin | None:
now = _utc_naive_now()
q = select(WatchPin).where(
WatchPin.admin_id == admin_id,
WatchPin.server_id == server_id,
WatchPin.expires_at > now,
)
return (await self.session.execute(q)).scalar_one_or_none()
pin = (await self.session.execute(q)).scalar_one_or_none()
if pin and _pin_still_occupied(pin, _utc_naive_now()):
return pin
return None
async def create_pin(
self,
@@ -90,7 +112,8 @@ class WatchRepositoryImpl:
ttl_hours: int = 8,
) -> dict[str, Any]:
now = _utc_naive_now()
expires = now + timedelta(hours=ttl_hours)
hours = normalize_watch_ttl_hours(ttl_hours)
expires = now + timedelta(hours=hours)
sess = WatchPinSession(
admin_id=admin_id,
server_id=server_id,
@@ -106,6 +129,8 @@ class WatchRepositoryImpl:
server_id=server_id,
pinned_at=now,
expires_at=expires,
monitoring_enabled=True,
ttl_hours=hours,
)
self.session.add(pin)
await self.session.flush()
@@ -116,6 +141,7 @@ class WatchRepositoryImpl:
"server_id": server_id,
"pinned_at": now,
"expires_at": expires,
"ttl_hours": hours,
}
async def end_session(self, session_id: int, reason: str) -> None:
@@ -159,34 +185,101 @@ class WatchRepositoryImpl:
ttl_hours=ttl_hours,
)
async def reset_pin_ttl(self, admin_id: int, server_id: int, ttl_hours: int = 8) -> dict[str, Any]:
"""Re-pin same server: end old session and start fresh 8h window."""
async def reset_pin_ttl(self, admin_id: int, server_id: int, ttl_hours: int | None = None) -> dict[str, Any]:
"""Re-pin same server: end old session and start fresh monitoring window."""
pin = await self.get_pin_by_server(admin_id, server_id)
if not pin:
raise ValueError("pin not found")
slot_index = pin.slot_index
hours = normalize_watch_ttl_hours(ttl_hours if ttl_hours is not None else int(pin.ttl_hours or 8))
await self.end_session(pin.session_id, "replaced")
await self.delete_pin(pin.id)
return await self.create_pin(
admin_id=admin_id,
server_id=server_id,
slot_index=slot_index,
ttl_hours=ttl_hours,
ttl_hours=hours,
)
async def remove_pin_by_slot(self, admin_id: int, slot_index: int) -> WatchPin | None:
return await self.delete_pin_by_slot(admin_id, slot_index)
async def count_monitoring_pins_for_server(self, server_id: int) -> int:
q = select(func.count()).select_from(WatchPin).where(
WatchPin.server_id == server_id,
WatchPin.monitoring_enabled.is_(True),
)
return int((await self.session.execute(q)).scalar_one() or 0)
async def _ensure_open_session(self, pin: WatchPin) -> None:
q = select(WatchPinSession).where(WatchPinSession.id == pin.session_id)
sess = (await self.session.execute(q)).scalar_one_or_none()
if sess and sess.ended_at is None:
return
now = _utc_naive_now()
new_sess = WatchPinSession(
admin_id=pin.admin_id,
server_id=pin.server_id,
slot_index=pin.slot_index,
started_at=now,
)
self.session.add(new_sess)
await self.session.flush()
pin.session_id = new_sess.id
async def set_monitoring_enabled(
self,
admin_id: int,
slot_index: int,
enabled: bool,
*,
ttl_hours: int | None = None,
) -> WatchPin | None:
pin = await self.get_pin_by_slot(admin_id, slot_index)
if not pin:
return None
if ttl_hours is not None:
pin.ttl_hours = normalize_watch_ttl_hours(ttl_hours)
if enabled:
await self._ensure_open_session(pin)
hours = normalize_watch_ttl_hours(int(pin.ttl_hours or 8))
pin.ttl_hours = hours
pin.monitoring_enabled = True
pin.expires_at = _utc_naive_now() + timedelta(hours=hours)
else:
pin.monitoring_enabled = False
await self.session.flush()
return pin
async def set_pin_ttl_hours(
self,
admin_id: int,
slot_index: int,
ttl_hours: int,
) -> WatchPin | None:
pin = await self.get_pin_by_slot(admin_id, slot_index)
if not pin:
return None
hours = normalize_watch_ttl_hours(ttl_hours)
pin.ttl_hours = hours
if pin.monitoring_enabled:
pin.expires_at = _utc_naive_now() + timedelta(hours=hours)
await self.session.flush()
return pin
async def expire_due_pins(self) -> list[int]:
now = _utc_naive_now()
q = select(WatchPin).where(WatchPin.expires_at <= now)
q = select(WatchPin).where(
WatchPin.expires_at <= now,
WatchPin.monitoring_enabled.is_(True),
)
pins = (await self.session.execute(q)).scalars().all()
session_ids: list[int] = []
server_ids: list[int] = []
for pin in pins:
await self.end_session(pin.session_id, "expired")
session_ids.append(pin.session_id)
await self.delete_pin(pin.id)
return session_ids
pin.monitoring_enabled = False
server_ids.append(pin.server_id)
return server_ids
async def insert_probe_record(self, **fields: Any) -> WatchProbeRecord:
row = WatchProbeRecord(**fields)
+111 -36
View File
@@ -1,13 +1,12 @@
"""SSH-based remote health / metrics probe (no Agent HTTP port required)."""
import json
import logging
from typing import Any
from server.domain.models import Server
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
logger = logging.getLogger("nexus.remote_probe")
from server.utils.watch_probe_errors import watch_probe_error_zh
# Emits one JSON line: {"status":"healthy","system_info":{...}}
SSH_HEALTH_METRICS_CMD = r"""python3 -c "
@@ -49,26 +48,45 @@ print(json.dumps({'status': 'healthy', 'system_info': info}))
SSH_WATCH_METRICS_CMD = r"""python3 -c "
import json, platform
import json
try:
import psutil
except ImportError:
print(json.dumps({'status':'error','error':'psutil missing'}))
raise SystemExit(0)
load = psutil.getloadavg()
net = psutil.net_io_counters()
disk_io = psutil.disk_io_counters()
try:
load = list(psutil.getloadavg())
except (AttributeError, OSError, NotImplementedError):
load = [0.0, 0.0, 0.0]
try:
net = psutil.net_io_counters()
except Exception:
net = None
try:
disk_io = psutil.disk_io_counters()
except Exception:
disk_io = None
try:
cpu = psutil.cpu_percent(interval=0.3)
mem = psutil.virtual_memory().percent
disk = round(psutil.disk_usage('/').percent, 1)
except Exception as exc:
print(json.dumps({'status':'error','error': str(exc)[:200]}))
raise SystemExit(0)
out = {
'status': 'ok',
'cpu_usage': psutil.cpu_percent(interval=0.3),
'mem_usage': psutil.virtual_memory().percent,
'disk_usage': round(psutil.disk_usage('/').percent, 1),
'cpu_usage': cpu,
'mem_usage': mem,
'disk_usage': disk,
'disk_mount': '/',
'load_avg': list(load),
'net_io': {'bytes_sent': net.bytes_sent, 'bytes_recv': net.bytes_recv},
'load_avg': load,
'net_io': {
'bytes_sent': int(net.bytes_sent) if net else 0,
'bytes_recv': int(net.bytes_recv) if net else 0,
},
'disk_io': {
'read_bytes': disk_io.read_bytes if disk_io else 0,
'write_bytes': disk_io.write_bytes if disk_io else 0,
'read_bytes': int(disk_io.read_bytes) if disk_io else 0,
'write_bytes': int(disk_io.write_bytes) if disk_io else 0,
},
}
print(json.dumps(out))
@@ -79,20 +97,39 @@ try:
except ImportError:
print(json.dumps({'status':'error','error':'psutil missing'}))
raise SystemExit(0)
load = psutil.getloadavg()
net = psutil.net_io_counters()
disk_io = psutil.disk_io_counters()
try:
load = list(psutil.getloadavg())
except (AttributeError, OSError, NotImplementedError):
load = [0.0, 0.0, 0.0]
try:
net = psutil.net_io_counters()
except Exception:
net = None
try:
disk_io = psutil.disk_io_counters()
except Exception:
disk_io = None
try:
cpu = psutil.cpu_percent(interval=0.3)
mem = psutil.virtual_memory().percent
disk = round(psutil.disk_usage('/').percent, 1)
except Exception as exc:
print(json.dumps({'status':'error','error': str(exc)[:200]}))
raise SystemExit(0)
out = {
'status': 'ok',
'cpu_usage': psutil.cpu_percent(interval=0.3),
'mem_usage': psutil.virtual_memory().percent,
'disk_usage': round(psutil.disk_usage('/').percent, 1),
'cpu_usage': cpu,
'mem_usage': mem,
'disk_usage': disk,
'disk_mount': '/',
'load_avg': list(load),
'net_io': {'bytes_sent': net.bytes_sent, 'bytes_recv': net.bytes_recv},
'load_avg': load,
'net_io': {
'bytes_sent': int(net.bytes_sent) if net else 0,
'bytes_recv': int(net.bytes_recv) if net else 0,
},
'disk_io': {
'read_bytes': disk_io.read_bytes if disk_io else 0,
'write_bytes': disk_io.write_bytes if disk_io else 0,
'read_bytes': int(disk_io.read_bytes) if disk_io else 0,
'write_bytes': int(disk_io.write_bytes) if disk_io else 0,
},
}
print(json.dumps(out))
@@ -125,16 +162,32 @@ print(json.dumps({'status':'ok','top_processes': procs[:5]}))
def _parse_json_stdout(stdout: str) -> dict[str, Any] | None:
for line in reversed(stdout.strip().splitlines()):
text = (stdout or "").strip()
if not text:
return None
candidates: list[str] = []
for line in reversed(text.splitlines()):
line = line.strip()
if not line.startswith("{"):
if not line:
continue
start = line.find("{")
if start >= 0:
candidates.append(line[start:])
elif line.startswith("{"):
candidates.append(line)
if text.startswith("{"):
candidates.append(text)
seen: set[str] = set()
for chunk in candidates:
if chunk in seen:
continue
seen.add(chunk)
try:
data = json.loads(line)
if isinstance(data, dict) and data.get("status"):
return data
data = json.loads(chunk)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and data.get("status"):
return data
return None
@@ -169,17 +222,39 @@ async def ssh_watch_probe(server: Server, timeout: int = 8) -> dict[str, Any]:
result = await exec_ssh_command(server, SSH_WATCH_METRICS_CMD, timeout=timeout)
duration_ms = int((time.monotonic() - started) * 1000)
if result.get("exit_code", -1) != 0:
err = (result.get("stderr") or result.get("stdout") or "SSH watch probe failed")[:300]
if "timed out" in err.lower() or duration_ms >= timeout * 1000 - 50:
return {"ok": False, "probe_status": "ssh_timeout", "error": err, "duration_ms": duration_ms}
if "auth" in err.lower() or "permission denied" in err.lower():
return {"ok": False, "probe_status": "ssh_auth", "error": err, "duration_ms": duration_ms}
return {"ok": False, "probe_status": "offline", "error": err, "duration_ms": duration_ms}
err_raw = (result.get("stderr") or result.get("stdout") or "SSH watch probe failed")[:300]
if "timed out" in err_raw.lower() or duration_ms >= timeout * 1000 - 50:
return {
"ok": False,
"probe_status": "ssh_timeout",
"error": watch_probe_error_zh(err_raw, status="ssh_timeout"),
"duration_ms": duration_ms,
}
if "auth" in err_raw.lower() or "permission denied" in err_raw.lower():
return {
"ok": False,
"probe_status": "ssh_auth",
"error": watch_probe_error_zh(err_raw, status="ssh_auth"),
"duration_ms": duration_ms,
}
return {
"ok": False,
"probe_status": "offline",
"error": watch_probe_error_zh(err_raw, status="offline"),
"duration_ms": duration_ms,
}
parsed = _parse_json_stdout(result.get("stdout", ""))
if not parsed or parsed.get("status") != "ok":
err = (parsed or {}).get("error") or "parse_error"
return {"ok": False, "probe_status": "parse_error", "error": str(err)[:255], "duration_ms": duration_ms}
raw = (result.get("stdout") or "").strip().replace("\n", " ")[:200]
err_key = (parsed or {}).get("error") or "parse_error"
err_combined = f"parse_error: {raw}" if err_key == "parse_error" and raw else str(err_key)
return {
"ok": False,
"probe_status": "parse_error",
"error": watch_probe_error_zh(err_combined, status="parse_error"),
"duration_ms": duration_ms,
}
parsed["duration_ms"] = duration_ms
return {"ok": True, "payload": parsed, "duration_ms": duration_ms}
+16
View File
@@ -10,6 +10,15 @@ from server.utils.fleet_metrics import _metric_from_system_info, _safe_pct
WATCH_MAX_SLOTS = 4
WATCH_PIN_TTL_HOURS = 8
WATCH_PIN_TTL_EXTENDED_HOURS = 24
WATCH_PIN_TTL_ALLOWED = frozenset({WATCH_PIN_TTL_HOURS, WATCH_PIN_TTL_EXTENDED_HOURS})
def normalize_watch_ttl_hours(hours: int) -> int:
"""Only 8h (default) or 24h (explicit opt-in) are allowed."""
if hours == WATCH_PIN_TTL_EXTENDED_HOURS:
return WATCH_PIN_TTL_EXTENDED_HOURS
return WATCH_PIN_TTL_HOURS
WATCH_PROBE_INTERVAL_SEC = 5
WATCH_REDIS_FRESH_SEC = 90
WATCH_PROBE_TIMEOUT_SEC = 8
@@ -84,6 +93,13 @@ def redis_sample_has_io(sample: WatchProbeSample | None) -> bool:
)
def redis_sample_has_basic_metrics(sample: WatchProbeSample | None) -> bool:
"""True when Agent heartbeat has CPU or memory — IO optional."""
if sample is None or sample.probe_status != PROBE_OK:
return False
return sample.cpu_pct is not None or sample.mem_pct is not None
def sanitize_processes(raw: Any) -> list[dict[str, Any]] | None:
if not isinstance(raw, list):
return None
+61
View File
@@ -0,0 +1,61 @@
"""Chinese user-facing messages for watch probe failures."""
from __future__ import annotations
_EXACT: dict[str, str] = {
"parse_error": "探针返回数据无法解析",
"psutil missing": "子机未安装 psutil(请执行 pip install psutil",
"ssh_probe_failed": "SSH 探针脚本执行失败",
"SSH watch probe failed": "SSH 监测探针失败",
"SSH probe failed": "SSH 探针失败",
"timeout": "连接超时",
"探针失败": "探针失败",
"无 Agent 且 SSH 探针失败": "无 Agent 且 SSH 探针失败",
"服务器不存在": "服务器不存在",
}
def watch_probe_error_zh(error: str | None, *, status: str | None = None) -> str | None:
"""Map probe error text to Chinese for UI / audit display."""
if not error:
if status == "parse_error":
return "探针返回数据无法解析"
if status == "ssh_timeout":
return "SSH 连接超时"
if status == "ssh_auth":
return "SSH 认证失败"
if status == "no_cred":
return "需要 Agent 或 SSH 凭据"
if status == "redis_stale":
return "Agent 心跳已过期"
if status == "offline":
return "目标服务器离线或不可达"
return None
text = error.strip()
if text in _EXACT:
return _EXACT[text]
lower = text.lower()
if lower.startswith("parse_error"):
detail = text.split(":", 1)[1].strip() if ":" in text else ""
if detail in _EXACT:
return f"探针返回数据无法解析:{_EXACT[detail]}"
if detail:
return f"探针返回数据无法解析:{detail}"
return "探针返回数据无法解析"
if "timed out" in lower or lower.endswith("timeout"):
return f"SSH 连接超时:{text}"
if "permission denied" in lower or "authentication failed" in lower:
return f"SSH 认证失败:{text}"
if "psutil missing" in lower:
return "子机未安装 psutil(请执行 pip install psutil"
if "ssh_probe_failed" in lower or "ssh watch probe failed" in lower:
return "SSH 探针脚本执行失败"
# Already Chinese or mixed — return as-is if contains CJK
if any("\u4e00" <= ch <= "\u9fff" for ch in text):
return text
return f"探针异常:{text}"
+19
View File
@@ -0,0 +1,19 @@
"""Watch slot Redis snapshots — cleared when no active 5s monitoring remains."""
from __future__ import annotations
from server.infrastructure.redis.client import get_redis
REDIS_WATCH_LIVE_PREFIX = "watch:live:"
REDIS_WATCH_PROC_PREFIX = "watch:proc:"
REDIS_WATCH_LAST_PREFIX = "watch:last:"
async def clear_watch_redis_snapshot(server_id: int) -> None:
"""Remove 5s watch sliding window; Agent continues normal 60s heartbeat."""
redis = get_redis()
await redis.delete(
f"{REDIS_WATCH_LIVE_PREFIX}{server_id}",
f"{REDIS_WATCH_PROC_PREFIX}{server_id}",
f"{REDIS_WATCH_LAST_PREFIX}{server_id}",
)
+17
View File
@@ -10,6 +10,7 @@ from server.utils.watch_metrics import (
merge_redis_and_ssh,
parse_ssh_watch_payload,
redis_sample_has_io,
redis_sample_has_basic_metrics,
sanitize_processes,
PROBE_OK,
)
@@ -44,6 +45,22 @@ def test_compute_rates_no_prev():
assert up is None and down is None and dr is None and dw is None
def test_redis_sample_has_basic_metrics():
basic = WatchProbeSample(probe_status=PROBE_OK, source="redis", cpu_pct=10)
assert redis_sample_has_basic_metrics(basic) is True
assert redis_sample_has_io(basic) is False
assert redis_sample_has_basic_metrics(None) is False
def test_parse_json_stdout_with_prefix_noise():
from server.infrastructure.ssh.remote_probe import _parse_json_stdout
stdout = "Warning: something\nOK {\"status\": \"ok\", \"cpu_usage\": 1.0}\n"
parsed = _parse_json_stdout(stdout)
assert parsed is not None
assert parsed.get("status") == "ok"
def test_redis_sample_has_io():
complete = WatchProbeSample(
probe_status=PROBE_OK,
+153
View File
@@ -17,7 +17,9 @@ def mock_watch_redis(monkeypatch):
mock_redis.lindex = AsyncMock(return_value=None)
mock_redis.lrange = AsyncMock(return_value=[])
mock_redis.get = AsyncMock(return_value=None)
mock_redis.delete = AsyncMock(return_value=1)
monkeypatch.setattr("server.application.services.watch_service.get_redis", lambda: mock_redis)
monkeypatch.setattr("server.utils.watch_state.get_redis", lambda: mock_redis)
return mock_redis
@@ -85,3 +87,154 @@ async def test_watch_probe_record_on_failure(db_session):
await db_session.commit()
assert row.probe_status == "ssh_timeout"
assert row.cpu_pct is None
@pytest.mark.asyncio
async def test_watch_pin_monitoring_toggle(db_session, test_admin, mock_watch_redis):
from server.application.services.watch_service import WatchService
server = Server(name="watch-toggle", domain="wt.example.com", port=22, agent_version="1.0")
db_session.add(server)
await db_session.commit()
svc = WatchService(db_session)
admin = test_admin["admin"]
await svc.pin_server(admin=admin, server_id=server.id)
listing = await svc.list_slots(admin.id)
assert listing["slots"][0]["monitoring_enabled"] is True
paused = await svc.set_slot_monitoring(
admin=admin,
slot_index=0,
monitoring_enabled=False,
)
assert paused["slots"][0]["monitoring_enabled"] is False
assert paused["slots"][0]["remaining_sec"] is None
resumed = await svc.set_slot_monitoring(
admin=admin,
slot_index=0,
monitoring_enabled=True,
)
assert resumed["slots"][0]["monitoring_enabled"] is True
assert resumed["slots"][0]["remaining_sec"] is not None
assert resumed["slots"][0]["remaining_sec"] > 0
@pytest.mark.asyncio
async def test_watch_pin_ttl_hours(db_session, test_admin, mock_watch_redis):
from server.application.services.watch_service import WatchService
from server.utils.watch_metrics import WATCH_PIN_TTL_EXTENDED_HOURS, WATCH_PIN_TTL_HOURS
server = Server(name="watch-ttl", domain="wt2.example.com", port=22, agent_version="1.0")
db_session.add(server)
await db_session.commit()
svc = WatchService(db_session)
admin = test_admin["admin"]
await svc.pin_server(admin=admin, server_id=server.id, ttl_hours=WATCH_PIN_TTL_HOURS)
listing = await svc.list_slots(admin.id)
assert listing["slots"][0]["ttl_hours"] == WATCH_PIN_TTL_HOURS
extended = await svc.set_slot_ttl(
admin=admin,
slot_index=0,
ttl_hours=WATCH_PIN_TTL_EXTENDED_HOURS,
)
assert extended["slots"][0]["ttl_hours"] == WATCH_PIN_TTL_EXTENDED_HOURS
assert extended["slots"][0]["remaining_sec"] >= WATCH_PIN_TTL_EXTENDED_HOURS * 3600 - 5
@pytest.mark.asyncio
async def test_watch_pin_ttl_expiry_list_syncs_before_display(db_session, test_admin, mock_watch_redis):
"""TTL 到期后、探针循环未跑 expire 前,list_slots 应同步为暂停态而非空槽。"""
from datetime import timedelta
from server.infrastructure.database.watch_repo import WatchRepositoryImpl, _utc_naive_now
from server.application.services.watch_service import WatchService
server = Server(name="watch-expire-list", domain="wel.example.com", port=22, agent_version="1.0")
db_session.add(server)
await db_session.commit()
svc = WatchService(db_session)
admin = test_admin["admin"]
await svc.pin_server(admin=admin, server_id=server.id)
repo = WatchRepositoryImpl(db_session)
pin = await repo.get_pin_by_slot(admin.id, 0)
assert pin is not None
pin.expires_at = _utc_naive_now() - timedelta(seconds=1)
await db_session.commit()
# 不手动调用 expire_due_pins — 模拟探针循环尚未处理的窗口
listing = await svc.list_slots(admin.id)
assert listing["slots"][0]["empty"] is False
assert listing["slots"][0]["monitoring_enabled"] is False
assert listing["slots"][0]["remaining_sec"] is None
@pytest.mark.asyncio
async def test_watch_pin_ttl_expiry_pin_server_no_duplicate(db_session, test_admin, mock_watch_redis):
"""到期 ghost pin 不应导致同槽 create_pin 唯一约束冲突。"""
from datetime import timedelta
from server.infrastructure.database.watch_repo import WatchRepositoryImpl, _utc_naive_now
from server.application.services.watch_service import WatchService
server_a = Server(name="watch-exp-a", domain="wea.example.com", port=22, agent_version="1.0")
server_b = Server(name="watch-exp-b", domain="web.example.com", port=22, agent_version="1.0")
db_session.add_all([server_a, server_b])
await db_session.commit()
svc = WatchService(db_session)
admin = test_admin["admin"]
await svc.pin_server(admin=admin, server_id=server_a.id, slot_index=0)
repo = WatchRepositoryImpl(db_session)
pin = await repo.get_pin_by_slot(admin.id, 0)
assert pin is not None
pin.expires_at = _utc_naive_now() - timedelta(seconds=1)
await db_session.commit()
# 替换槽位服务器 — 应先 finalize 再 replace,不能 IntegrityError
result = await svc.pin_server(admin=admin, server_id=server_b.id, slot_index=0)
assert result["slots"][0]["empty"] is False
assert result["slots"][0]["server_id"] == server_b.id
assert result["slots"][0]["monitoring_enabled"] is True
@pytest.mark.asyncio
async def test_watch_pin_ttl_expiry_pauses_not_deletes(db_session, test_admin, mock_watch_redis):
from datetime import timedelta
from server.infrastructure.database.watch_repo import WatchRepositoryImpl, _utc_naive_now
from server.application.services.watch_service import WatchService
server = Server(name="watch-expire", domain="we.example.com", port=22, agent_version="1.0")
db_session.add(server)
await db_session.commit()
svc = WatchService(db_session)
admin = test_admin["admin"]
await svc.pin_server(admin=admin, server_id=server.id)
repo = WatchRepositoryImpl(db_session)
pin = await repo.get_pin_by_slot(admin.id, 0)
assert pin is not None
pin.expires_at = _utc_naive_now() - timedelta(seconds=1)
await db_session.commit()
expired = await repo.expire_due_pins()
assert server.id in expired
await db_session.commit()
still = await repo.get_pin_by_slot(admin.id, 0)
assert still is not None
assert still.monitoring_enabled is False
listing = await svc.list_slots(admin.id)
assert listing["slots"][0]["empty"] is False
assert listing["slots"][0]["monitoring_enabled"] is False
+23
View File
@@ -0,0 +1,23 @@
"""Tests for watch probe error localization."""
from server.utils.watch_probe_errors import watch_probe_error_zh
def test_parse_error_zh():
assert watch_probe_error_zh("parse_error") == "探针返回数据无法解析"
assert watch_probe_error_zh("psutil missing") == "子机未安装 psutil(请执行 pip install psutil"
def test_parse_error_with_detail():
out = watch_probe_error_zh("parse_error: psutil missing")
assert "探针返回数据无法解析" in out
assert "psutil" in out
def test_timeout_zh():
out = watch_probe_error_zh("Connection timed out during banner exchange")
assert out.startswith("SSH 连接超时")
def test_status_only_fallback():
assert watch_probe_error_zh(None, status="ssh_timeout") == "SSH 连接超时"