From 765d73b34286ed9beec06f8eb299fcb886ee4c94 Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Sat, 18 Jul 2026 20:10:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(observability):=20worker=20=E5=BF=83?= =?UTF-8?q?=E8=B7=B3=E5=91=8A=E8=AD=A6=E9=97=AD=E7=8E=AF=20+=20topic=20las?= =?UTF-8?q?t=5Ferror=20=E9=9D=A2=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 worker 故障只记日志：心跳过期只触发 docker restart（无人工告警）， topic 抓取失败静默（last_error 被 API 用 CollectionAction.created_at 覆盖掩盖）。运维只能登服务器看日志发现问题。 后端： - apps/worker/main.py：心跳改写共享卷 /app/data/worker_heartbeat.json （pm_data 卷，backend 可读）+ 新增 heartbeat_alert_job（每10min自检）： 心跳过期/主题 last_error 非空 → 发告警邮件给 notify_default_to， 带 1h 去重防刷屏（模块级 _last_alerts）。注册到 scheduler（*/10）。 - scripts/worker_healthcheck.py：改读共享卷心跳（与 worker 写入对齐）。 - apps/api/routers/topics.py：list_topics/_topic_dict 修复 last_run_at/ last_error 掩盖——改读 TopicSubscription 真实抓取状态（PR1 存的）， CollectionAction.created_at 改作 last_action_at 单独字段。加 ?failed=true 过滤（只返回 last_error 非空的 topic，供面板）。 - apps/api/routers/system.py：新增 /system/worker 端点（读共享卷心跳返 {ts,error,age_seconds,is_stale}）；system_status 增补 worker_heartbeat + topic_errors 计数。 前端： - types/index.ts：Topic 加 last_error/last_action_at；新增 WorkerHeartbeat 类型；SystemStatus 加 worker_heartbeat/topic_errors。 - services/api.ts：systemApi.worker + topicApi.list(failed) + WorkerHeartbeat 导入。 - pages/Operations.tsx：加 Worker 心跳面板（绿/红时效判定）+ 主题抓取错误 面板（列表 last_error + 重新抓取按钮复用 topicApi.fetch）。mount 时加载。 测试：新增 TestWorkerAlertDedup（去重+无收件人静默）、TestTopicLastErrorExposed （last_error 持久化+成功清空），全套 61 passed。tsc --noEmit 通过。 --- apps/api/routers/system.py | 39 +++++++++ apps/api/routers/topics.py | 23 +++-- apps/worker/main.py | 101 +++++++++++++++++++++- frontend/src/pages/Operations.tsx | 137 +++++++++++++++++++++++++++++- frontend/src/services/api.ts | 5 +- frontend/src/types/index.ts | 11 +++ scripts/worker_healthcheck.py | 8 +- tests/test_repositories.py | 69 +++++++++++++++ 8 files changed, 379 insertions(+), 14 deletions(-) diff --git a/apps/api/routers/system.py b/apps/api/routers/system.py index 72a2d38..49d1afd 100644 --- a/apps/api/routers/system.py +++ b/apps/api/routers/system.py @@ -2,6 +2,9 @@ @author Color2333 """ +import time +from pathlib import Path + from fastapi import APIRouter, Query from apps.api.deps import iso_dt, settings @@ -15,6 +18,28 @@ router = APIRouter() +# worker 心跳文件（pm_data 共享卷，worker 写、backend 读） +_WORKER_HEARTBEAT_FILE = Path("/app/data/worker_heartbeat.json") +_HEARTBEAT_STALE_SECONDS = 1200 # 与 worker healthcheck 一致 + + +def _read_worker_heartbeat() -> dict: + """读共享卷 worker 心跳，返回 {ts, error, age_seconds, is_stale}；文件缺失返回 None。""" + import json + + try: + data = json.loads(_WORKER_HEARTBEAT_FILE.read_text()) + ts = float(data.get("ts", 0)) + age = time.time() - ts + return { + "ts": ts, + "error": data.get("error"), + "age_seconds": int(age), + "is_stale": age > _HEARTBEAT_STALE_SECONDS, + } + except (OSError, ValueError, TypeError): + return None + @router.get("/health") def health() -> dict: @@ -28,6 +53,16 @@ def health() -> dict: } +@router.get("/system/worker") +def worker_status() -> dict: + """Worker 心跳状态（可观测性：供 Operations 面板 + 告警自检查询）""" + hb = _read_worker_heartbeat() + return { + "heartbeat": hb, + "stale_threshold": _HEARTBEAT_STALE_SECONDS, + } + + @router.get("/system/status") def system_status() -> dict: with session_scope() as session: @@ -36,6 +71,8 @@ def system_status() -> dict: papers_total = PaperRepository(session).count_all() runs = PipelineRunRepository(session).list_latest(limit=50) failed = [r for r in runs if r.status.value == "failed"] + # 可观测性：附带 worker 心跳摘要 + 主题抓取错误计数 + errored_topics = [t for t in topics if t.last_error] return { "health": health(), "counts": { @@ -45,6 +82,8 @@ def system_status() -> dict: "runs_latest_50": len(runs), "failed_runs_latest_50": len(failed), }, + "worker_heartbeat": _read_worker_heartbeat(), + "topic_errors": len(errored_topics), "latest_run": ( { "pipeline_name": runs[0].pipeline_name, diff --git a/apps/api/routers/topics.py b/apps/api/routers/topics.py index 8f7d121..f451591 100644 --- a/apps/api/routers/topics.py +++ b/apps/api/routers/topics.py @@ -36,8 +36,12 @@ def _topic_dict(t, session=None) -> dict: "enable_date_filter": getattr(t, "enable_date_filter", False), "date_filter_days": getattr(t, "date_filter_days", 7), "paper_count": 0, - "last_run_at": None, + # last_run_at/last_error 读 TopicSubscription 真实抓取状态（PR1 存的）， + # 此前被 CollectionAction.created_at 覆盖掩盖了抓取失败 + "last_run_at": t.last_run_at.isoformat() if t.last_run_at else None, + "last_error": t.last_error, "last_run_count": None, + "last_action_at": None, # 最近一次收集行动（与 last_run_at 区分） } if session is not None: from sqlalchemy import func, select @@ -49,7 +53,7 @@ def _topic_dict(t, session=None) -> dict: select(func.count()).select_from(PaperTopic).where(PaperTopic.topic_id == t.id) ) d["paper_count"] = cnt or 0 - # 最近一次行动 + # 最近一次收集行动（单独字段，不再覆盖 last_run_at） last_action = session.execute( select(CollectionAction) .where(CollectionAction.topic_id == t.id) @@ -57,7 +61,7 @@ def _topic_dict(t, session=None) -> dict: .limit(1) ).scalar_one_or_none() if last_action: - d["last_run_at"] = ( + d["last_action_at"] = ( last_action.created_at.isoformat() if last_action.created_at else None ) d["last_run_count"] = last_action.paper_count @@ -65,13 +69,16 @@ def _topic_dict(t, session=None) -> dict: @router.get("/topics") -def list_topics(enabled_only: bool = False) -> dict: +def list_topics(enabled_only: bool = False, failed: bool = False) -> dict: from sqlalchemy import func, select from packages.storage.models import CollectionAction, PaperTopic with session_scope() as session: topics = TopicRepository(session).list_topics(enabled_only=enabled_only) + # failed=true：只返回最近抓取出错的 topic（供可观测性面板） + if failed: + topics = [t for t in topics if t.last_error] if not topics: return {"items": []} topic_ids = [t.id for t in topics] @@ -110,12 +117,16 @@ def list_topics(enabled_only: bool = False) -> dict: "enabled": t.enabled, "created_at": t.created_at.isoformat() if t.created_at else None, "paper_count": paper_counts.get(t.id, 0), - "last_run_at": None, + # last_run_at/last_error 读 TopicSubscription 真实抓取状态（PR1）， + # 此前被 CollectionAction.created_at 覆盖，抓取失败被掩盖 + "last_run_at": t.last_run_at.isoformat() if t.last_run_at else None, + "last_error": t.last_error, + "last_action_at": None, "last_run_count": None, } last_action = latest_actions.get(t.id) if last_action: - d["last_run_at"] = ( + d["last_action_at"] = ( last_action.created_at.isoformat() if last_action.created_at else None ) d["last_run_count"] = last_action.paper_count diff --git a/apps/worker/main.py b/apps/worker/main.py index 47f650c..584f85c 100644 --- a/apps/worker/main.py +++ b/apps/worker/main.py @@ -36,9 +36,13 @@ setup_logging() logger = logging.getLogger(__name__) -_HEALTH_FILE = Path("/tmp/worker_heartbeat") +# 心跳改写共享卷 pm_data（/app/data），backend 也能读同一文件暴露 worker 状态。 +# 此前写 /tmp（容器内，后端读不到），可观测性端点无法查询 worker 健康。 +_HEALTH_FILE = Path("/app/data/worker_heartbeat.json") # 心跳健康判定：最近一次心跳距现在超过此秒数视为不健康（捕获 worker 卡死/全部任务失败） _HEARTBEAT_STALE_SECONDS = 1200 # 20 分钟（cron job 最小间隔 30min，留足缓冲） +# 告警去重：同一告警类型在此秒数内不重复发邮件（防刷屏） +_ALERT_DEDUP_SECONDS = 3600 # 1 小时 def _write_heartbeat(error: str | None = None) -> None: @@ -56,6 +60,92 @@ def _write_heartbeat(error: str | None = None) -> None: ) +def _read_heartbeat() -> dict | None: + """读共享卷心跳文件，供自检告警 + 端点查询。文件缺失/损坏返回 None。""" + import json + + try: + return json.loads(_HEALTH_FILE.read_text()) + except (OSError, ValueError, TypeError): + return None + + +# 告警去重状态：{alert_key: last_alert_ts}（进程内，重启后重置——可接受） +_last_alerts: dict[str, float] = {} + + +def _send_alert(subject: str, html: str, alert_key: str) -> None: + """发告警邮件给 notify_default_to，带 1h 去重。SMTP 未配置时静默跳过。""" + now = time.time() + last = _last_alerts.get(alert_key, 0) + if now - last < _ALERT_DEDUP_SECONDS: + logger.debug( + "告警 %s 去重中（距上次 %.0fs < %ds），跳过", + alert_key, + now - last, + _ALERT_DEDUP_SECONDS, + ) + return + from packages.config import get_settings + from packages.integrations.notifier import NotificationService + + recipient = get_settings().notify_default_to + if not recipient: + logger.debug("notify_default_to 未配置，跳过告警 %s", alert_key) + return + ok = NotificationService().send_email_html(recipient, subject, html) + if ok: + _last_alerts[alert_key] = now + logger.info("告警邮件已发送: %s -> %s", alert_key, recipient) + else: + logger.warning("告警邮件发送失败（SMTP 未配置或出错）: %s", alert_key) + + +def heartbeat_alert_job() -> None: + """每 10min 自检：心跳过期 + 主题抓取错误 → 发告警邮件（可观测性闭环）。 + + 心跳过期说明 worker 卡死或全部 job 失败；主题 last_error 说明某主题抓取失败。 + 两者此前只记日志无人知，现在发邮件给 notify_default_to（带 1h 去重防刷屏）。 + """ + import html as html_lib + + # 1. 心跳过期检查 + hb = _read_heartbeat() + if hb is None or (time.time() - float(hb.get("ts", 0))) > _HEARTBEAT_STALE_SECONDS: + age = ( + "未知（文件缺失/损坏）" + if hb is None + else f"{int(time.time() - float(hb.get('ts', 0)))}s" + ) + body = ( + f"

⚠️ Worker 心跳过期

" + f"

心跳距今 {age}（阈值 {_HEARTBEAT_STALE_SECONDS}s）。

" + f"

可能原因：worker 卡死、全部 job 失败、或容器异常。

" + f"

最近错误：{(hb or {}).get('error') or 'N/A'}

" + f"

请检查 worker 容器状态与日志。

" + ) + _send_alert("[PaperMind] Worker 心跳过期告警", body, "heartbeat_stale") + # 2. 主题抓取错误检查 + try: + with session_scope() as session: + topics = TopicRepository(session).list_topics(enabled_only=True) + errored = [t for t in topics if t.last_error] + if errored: + rows = "".join( + f"{html_lib.escape(t.name)}" + f"{html_lib.escape((t.last_error or '')[:200])}" + for t in errored + ) + body = ( + f"

⚠️ 主题抓取错误（{len(errored)} 个）

" + f"" + f"{rows}
主题最近错误
" + ) + _send_alert(f"[PaperMind] {len(errored)} 个主题抓取失败", body, "topic_errors") + except Exception: + logger.exception("heartbeat_alert_job 检查主题错误失败") + + def _update_topic_run_status(topic_id: str, *, error: str | None) -> None: """记录主题抓取的最近运行时间与错误（Critical #4：失败可查可补抓）。 @@ -304,6 +394,15 @@ def run_worker() -> None: ) logger.info("✅ 已添加：每周图谱维护任务（UTC 周日 22:00）") + # 可观测性：心跳过期 + 主题抓取错误告警（每 10min 自检发邮件） + scheduler.add_job( + heartbeat_alert_job, + trigger=CronTrigger(minute="*/10"), + id="heartbeat_alert", + **_job_kwargs, + ) + logger.info("✅ 已添加：心跳告警自检任务（每 10min）") + # 优雅关闭（High 3f：等待进行中任务跑完，避免已下载 PDF 未 set_pdf_path # 的中间态丢失；wait=True + 60s 超时兜底） def _graceful_stop(*_: object) -> None: diff --git a/frontend/src/pages/Operations.tsx b/frontend/src/pages/Operations.tsx index 234a04f..bb529af 100644 --- a/frontend/src/pages/Operations.tsx +++ b/frontend/src/pages/Operations.tsx @@ -3,11 +3,11 @@ * 覆盖 API: POST /citations/sync/*, POST /jobs/*, GET /system/status * @author Color2333 */ -import { useState } from "react"; +import { useEffect, useState } from "react"; import { Card, CardHeader, Button, Input } from "@/components/ui"; import { useToast } from "@/contexts/ToastContext"; -import { citationApi, jobApi, systemApi } from "@/services/api"; -import type { CitationSyncResult, SystemStatus } from "@/types"; +import { citationApi, jobApi, systemApi, topicApi } from "@/services/api"; +import type { CitationSyncResult, SystemStatus, Topic, WorkerHeartbeat } from "@/types"; import { Settings, Link2, @@ -18,6 +18,8 @@ import { Network, Calendar, Zap, + HeartPulse, + RotateCw, } from "lucide-react"; interface OperationResult { @@ -35,6 +37,46 @@ export default function Operations() { const [syncPaperId, setSyncPaperId] = useState(""); /* 引用同步 - 主题 */ const [syncTopicId, setSyncTopicId] = useState(""); + /* 可观测性：worker 心跳 + 主题抓取错误 */ + const [heartbeat, setHeartbeat] = useState(null); + const [staleThreshold, setStaleThreshold] = useState(1200); + const [failedTopics, setFailedTopics] = useState([]); + const [obsLoading, setObsLoading] = useState(false); + const [refetchingId, setRefetchingId] = useState(null); + + const refreshObservability = async () => { + setObsLoading(true); + try { + const [workerRes, topicsRes] = await Promise.all([ + systemApi.worker(), + topicApi.list(false, true), + ]); + setHeartbeat(workerRes.heartbeat); + setStaleThreshold(workerRes.stale_threshold); + setFailedTopics(topicsRes.items); + } catch (err) { + toast("error", err instanceof Error ? err.message : "可观测性数据加载失败"); + } finally { + setObsLoading(false); + } + }; + + useEffect(() => { + refreshObservability(); + }, []); + + const handleRefetchTopic = async (id: string, name: string) => { + setRefetchingId(id); + try { + await topicApi.fetch(id); + toast("success", `✅ 主题「${name}」重新抓取已启动`); + refreshObservability(); + } catch (err) { + toast("error", err instanceof Error ? err.message : "重新抓取失败"); + } finally { + setRefetchingId(null); + } + }; const setLoading = (key: string, val: boolean) => setLoadings((prev) => ({ ...prev, [key]: val })); @@ -321,6 +363,95 @@ export default function Operations() { + + {/* 可观测性：Worker 心跳 + 主题抓取错误 */} +
+ {/* Worker 心跳 */} + + + + + } + /> +
+ {heartbeat === null ? ( +
+ + 心跳文件缺失或损坏 — worker 可能未启动或共享卷未挂载 +
+ ) : heartbeat.is_stale ? ( +
+ +
+

心跳过期（{heartbeat.age_seconds}s > {staleThreshold}s）

+

worker 可能卡死或全部任务失败。最近错误：{heartbeat.error || "N/A"}

+

worker 每 10min 自检发告警邮件给 notify_default_to。

+
+
+ ) : ( +
+ + 正常 — 心跳距今 {heartbeat.age_seconds}s（阈值 {staleThreshold}s） +
+ )} +
+ + + {/* 主题抓取错误 */} + + } + /> +
+ {failedTopics.length === 0 ? ( +
+ + 无抓取错误的 topic +
+ ) : ( +
+ {failedTopics.map((t) => ( +
+
+ {t.name} + +
+

+ {t.last_run_at ? `最近运行：${new Date(t.last_run_at).toLocaleString()}` : "无运行记录"} +

+

+ {t.last_error || "未知错误"} +

+
+ ))} +
+ )} +
+ +
); } diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index 3785f1f..b01c042 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -5,6 +5,7 @@ import { retryAsync } from "@/lib/errorHandler"; import type { SystemStatus, + WorkerHeartbeat, Topic, TopicCreate, TopicUpdate, @@ -231,6 +232,7 @@ function del(path: string, opts?: { signal?: AbortSignal }) { export const systemApi = { health: () => get<{ status: string; app: string; env: string }>("/health"), status: () => get("/system/status"), + worker: () => get<{ heartbeat: WorkerHeartbeat | null; stale_threshold: number }>("/system/worker"), }; export const todayApi = { @@ -239,7 +241,8 @@ export const todayApi = { /* ========== 主题 ========== */ export const topicApi = { - list: (enabledOnly = false) => get<{ items: Topic[] }>(`/topics?enabled_only=${enabledOnly}`), + list: (enabledOnly = false, failed = false) => + get<{ items: Topic[] }>(`/topics?enabled_only=${enabledOnly}${failed ? "&failed=true" : ""}`), create: (data: TopicCreate) => post("/topics", data), update: (id: string, data: TopicUpdate) => patch(`/topics/${id}`, data), delete: (id: string) => del<{ deleted: string }>(`/topics/${id}`), diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 3501746..c3ca97e 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -19,9 +19,18 @@ export interface SystemStatus { runs_latest_50: number; failed_runs_latest_50: number; }; + worker_heartbeat?: WorkerHeartbeat | null; + topic_errors?: number; latest_run: PipelineRun | null; } +export interface WorkerHeartbeat { + ts: number; + error: string | null; + age_seconds: number; + is_stale: boolean; +} + /* ========== 主题 ========== */ export type ScheduleFrequency = "daily" | "twice_daily" | "weekdays" | "weekly"; @@ -38,6 +47,8 @@ export interface Topic { date_filter_days: number; paper_count?: number; last_run_at?: string | null; + last_error?: string | null; + last_action_at?: string | null; last_run_count?: number | null; } diff --git a/scripts/worker_healthcheck.py b/scripts/worker_healthcheck.py index f06803f..39af10a 100644 --- a/scripts/worker_healthcheck.py +++ b/scripts/worker_healthcheck.py @@ -1,12 +1,14 @@ #!/usr/bin/env python3 -"""Worker 健康检查脚本（High 2e）。 +"""Worker 健康检查脚本（High 2e + 可观测性：改读共享卷）。 -读取 /tmp/worker_heartbeat（JSON {ts, error}），判定心跳时效： +读取 /app/data/worker_heartbeat.json（pm_data 共享卷，backend 也可读）， +判定心跳时效： - 文件不存在 / 解析失败 → 不健康（exit 1） - ts 距今超过 1200 秒 → 不健康（worker 卡死或全部任务失败，心跳已过期） - 否则健康（exit 0） 此前 healthcheck 仅 test -f 文件存在 → 即使所有 job 失败 worker 仍判健康。 +心跳改共享卷后，backend /system/worker 端点也能读同一文件暴露 worker 状态。 """ from __future__ import annotations @@ -16,7 +18,7 @@ import time from pathlib import Path -HEALTH_FILE = Path("/tmp/worker_heartbeat") +HEALTH_FILE = Path("/app/data/worker_heartbeat.json") STALE_SECONDS = 1200 # 20 分钟 diff --git a/tests/test_repositories.py b/tests/test_repositories.py index ae00fd4..8c9bc6e 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -424,3 +424,72 @@ def test_link_category_isolation(self, db_session): assert len(PaperRepository(db_session).list_by_topic(t_ai.id)) == 1 assert len(PaperRepository(db_session).list_by_topic(t_lg.id)) == 1 assert t_ai.id != t_lg.id + + +class TestWorkerAlertDedup: + """Worker 告警去重逻辑测试（可观测性：同错误 1h 内不重复发邮件）""" + + def test_send_alert_dedup_within_window(self, monkeypatch): + from apps.worker import main as wm + + # 重置去重状态 + wm._last_alerts.clear() + sent: list[str] = [] + + def fake_send(self, recipient, subject, html): + sent.append(subject) + return True + + monkeypatch.setattr( + "packages.integrations.notifier.NotificationService.send_email_html", fake_send + ) + monkeypatch.setattr( + "packages.config.get_settings", + lambda: type("S", (), {"notify_default_to": "test@example.com"})(), + ) + + wm._send_alert("[PaperMind] Test Alert", "

x

", "test_key") + assert len(sent) == 1, "首次告警应发送" + # 窗口内再次告警 → 去重，不发送 + wm._send_alert("[PaperMind] Test Alert", "

x

", "test_key") + assert len(sent) == 1, "去重窗口内不应重复发送" + wm._last_alerts.clear() + + def test_send_alert_no_recipient_silent(self, monkeypatch): + from apps.worker import main as wm + + wm._last_alerts.clear() + sent: list[str] = [] + + def fake_send(self, recipient, subject, html): + sent.append(subject) + return True + + monkeypatch.setattr( + "packages.integrations.notifier.NotificationService.send_email_html", fake_send + ) + # notify_default_to 未配置 + monkeypatch.setattr( + "packages.config.get_settings", + lambda: type("S", (), {"notify_default_to": None})(), + ) + wm._send_alert("[PaperMind] Test", "

x

", "no_recv") + assert len(sent) == 0, "无收件人应静默跳过" + wm._last_alerts.clear() + + +class TestTopicLastErrorExposed: + """主题 last_error 经 repository 可读（可观测性：API 不再掩盖）""" + + def test_last_error_persisted_and_readable(self, db_session): + repo = TopicRepository(db_session) + topic = repo.upsert_topic(name="ErrTopic", query="q") + db_session.flush() + repo.update_run_status(topic.id, error="arxiv 429 限流") + db_session.refresh(topic) + assert topic.last_error == "arxiv 429 限流" + assert topic.last_run_at is not None + # 成功后清空 + repo.update_run_status(topic.id, error=None) + db_session.refresh(topic) + assert topic.last_error is None