refactor: 移除日志实时流功能
删除 log_stream 模块及 LOG_TAIL_CTRL 命令、懒加载与停机释放逻辑，日志查看仅保留 LOG_TAIL 与 GET_LOG_DOWNLOAD。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Code
+1
-20
@@ -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 触发主程序优雅重启
|
||||
|
||||
@@ -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()),
|
||||
})
|
||||
+1
-3
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user