From a6a801220c24467d08874d9ce92fea4cc75a6709 Mon Sep 17 00:00:00 2001 From: lianlonggang Date: Wed, 9 Sep 2026 14:32:29 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E7=A7=BB=E9=99=A4=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E5=AE=9E=E6=97=B6=E6=B5=81=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 删除 log_stream 模块及 LOG_TAIL_CTRL 命令、懒加载与停机释放逻辑，日志查看仅保留 LOG_TAIL 与 GET_LOG_DOWNLOAD。 Co-Authored-By: Claude Code --- agent/dispatcher.py | 21 +------ agent/log_stream.py | 141 -------------------------------------------- agent/main.py | 4 +- 3 files changed, 2 insertions(+), 164 deletions(-) delete mode 100644 agent/log_stream.py diff --git a/agent/dispatcher.py b/agent/dispatcher.py index b2a191f..b1638b1 100644 --- a/agent/dispatcher.py +++ b/agent/dispatcher.py @@ -9,7 +9,7 @@ B 端通过 WebSocket 下发 JSON 命令：{"type":"CMD_TYPE","payload":{...}} - GET_VERSION 查询主程序版本 - GET_LOG_FILES 日志文件列表 - LOG_TAIL 日志尾部（payload: file, lines） -- LOG_TAIL_CTRL 日志实时流开关（payload: action, file） +- GET_LOG_DOWNLOAD 日志全量下载（payload: file） - RESTART 重启主程序（优雅停机后由 ProcessGuard 拉起） - GET_CONFIG 读取医院配置文件（payload: hospital） - SAVE_CONFIG 写回医院配置文件（payload: hospital, content） @@ -36,8 +36,6 @@ class CommandDispatcher: self._client = None # 由 main 注入（避免构造环） self._upgrade_command = None # 升级命令（懒加载） - self._log_stream = None # 日志实时流管理（懒加载） - self._log_stream_lock = threading.Lock() self._upgrade_lock = threading.Lock() def set_client(self, client): @@ -52,18 +50,6 @@ class CommandDispatcher: self._upgrade_command = UpgradeCommand(self.config, self.main_app_client, self._client) return self._upgrade_command - def _get_log_stream(self): - """懒加载日志流管理器。""" - with self._log_stream_lock: - if self._log_stream is None: - from agent.log_stream import LogStreamManager - self._log_stream = LogStreamManager(self.config, self.main_app_client, self._client) - return self._log_stream - - def shutdown_log_stream(self): - """日志实时流（供 main 停机时统一释放）。""" - self._get_log_stream().stop_all() - # ================= 分发 ================= def dispatch(self, msg_type, payload): @@ -89,11 +75,6 @@ class CommandDispatcher: lines = (payload or {}).get("lines") or 200 result = {"ok": True, "data": self.main_app_client.log_tail(file, lines)} - elif msg_type == "LOG_TAIL_CTRL": - action = (payload or {}).get("action") or "start" - file = (payload or {}).get("file") or "info.log" - result = self._get_log_stream().control(action, file) - elif msg_type == "RESTART": # 优先走 NSSM 服务重启（主程序假死时 HTTP 不可达也能重启成功）； # 未配置服务名时回退环回 HTTP 触发主程序优雅重启 diff --git a/agent/log_stream.py b/agent/log_stream.py deleted file mode 100644 index 323e8e0..0000000 --- a/agent/log_stream.py +++ /dev/null @@ -1,141 +0,0 @@ -# -*- coding: utf-8 -*- -"""日志实时推送管理器。 - -B 端通过 LOG_TAIL_CTRL 命令开启/停止某设备某日志文件的实时流： -- 开启后每 2 秒从主程序 /api/admin/log/tail 拉取增量行 -- 新行打包为 LOG_BATCH 上报 B 端（B 端转发给浏览器展示） -- 每个文件独立游标（已发送行数），支持多文件并行 - -增量算法：tail(max(offset+FETCH, 400)) 行，取 offset 之后的部分； -日志滚动（文件变小/重置）时自动重置游标。 -""" - -import json -import logging -import threading -import time - -log = logging.getLogger(__name__) - -# 轮询间隔（秒） -POLL_SEC = 2 -# 单次拉取上限（行） -FETCH_MAX = 400 - - -class LogStreamManager: - """日志实时流：单线程轮询全部活跃文件，增量上报 LOG_BATCH。""" - - def __init__(self, config, main_app_client, client): - self.config = config - self.main_app_client = main_app_client - self.client = client - - self._cursors: dict[str, int] = {} # 文件名 → 已发送行数 - self._active: dict[str, bool] = {} # 文件名 → 是否激活 - self._lock = threading.Lock() - self._stop = threading.Event() - self._thread: threading.Thread | None = None - - # ================= 控制入口 ================= - - def control(self, action: str, file: str) -> dict: - """ - 控制命令入口。 - - :param action: start / stop - :param file: 日志文件名（info.log / error.log） - """ - if action == "start": - with self._lock: - if self._active.get(file): - return {"ok": True, "msg": f"流已存在：{file}"} - self._active[file] = True - self._cursors.pop(file, None) - self._ensure_thread() - log.info("日志流开启：%s", file) - return {"ok": True, "msg": f"日志流已开启：{file}"} - - if action == "stop": - with self._lock: - self._active.pop(file, None) - self._cursors.pop(file, None) - log.info("日志流停止：%s", file) - return {"ok": True, "msg": f"日志流已停止：{file}"} - - return {"ok": False, "msg": f"未知 action：{action}（支持 start/stop）"} - - def stop_all(self): - """停止全部流（Agent 停机时调用）。""" - self._stop.set() - with self._lock: - self._active.clear() - self._cursors.clear() - - def _ensure_thread(self): - """确保轮询线程在运行（首个流开启时启动）。""" - if self._thread is not None and self._thread.is_alive(): - return - self._thread = threading.Thread(target=self._poll_loop, name="log-stream", daemon=True) - self._thread.start() - - # ================= 轮询 ================= - - def _poll_loop(self): - """轮询主循环：每 POLL_SEC 秒拉一轮全部活跃文件。""" - while not self._stop.wait(POLL_SEC): - with self._lock: - files = list(self._active.keys()) - for file in files: - if self._stop.is_set(): - return - try: - self._poll_one(file) - except Exception as e: - log.warning("日志流轮询失败（%s）：%s", file, e) - - def _poll_one(self, file: str): - """单轮增量拉取：取尾部行，比对游标，新行打包上报。""" - with self._lock: - if not self._active.get(file): - return - offset = self._cursors.get(file, 0) - - # 多拉一段，保证能覆盖到 offset 之前的已有内容 - fetch = max(offset + 100, FETCH_MAX) - raw = self.main_app_client.log_tail(file, fetch) - log.debug("日志流轮询：file=%s, offset=%s, fetch=%s, rawLen=%s", - file, offset, fetch, len(raw) if raw else -1) - if raw is None: - return - try: - node = json.loads(raw) - except Exception: - return - if node.get("code") != 0: - return - content = node.get("data") - if not content: - return - - all_lines = content.split("\n") - - # 日志滚动检测：行数比游标还少很多 → 文件已重置 - if len(all_lines) < offset: - with self._lock: - self._cursors[file] = 0 - return - if len(all_lines) <= offset: - return # 无新行 - - fresh = [line for line in all_lines[offset:] if line and line.strip()] - with self._lock: - self._cursors[file] = len(all_lines) - if not fresh: - return - - self.client.send("LOG_BATCH", { - "file": file, - "lines": fresh, - "ts": int(time.time()), - }) diff --git a/agent/main.py b/agent/main.py index 6b1b058..2f21fe6 100644 --- a/agent/main.py +++ b/agent/main.py @@ -89,10 +89,8 @@ def _setup_logging(log_dir: str): def _shutdown(signum, frame): - """优雅停机：释放日志流 → 停守护 → 断连接 → 退出。""" + """优雅停机：停守护 → 断连接 → 退出。""" log.info("Agent 停机中...（signal=%s）", signum) - if _dispatcher is not None: - _dispatcher.shutdown_log_stream() if _guard is not None: _guard.stop() if _client is not None: