lianlonggangandClaude
68751c2354
- WebSocket 连接 B 端（注册/心跳/指数退避重连） - 命令分发（10 种远程运维命令） - 看门狗守护、日志实时流、在线升级含自动回滚 - 修复 main.py 日志格式 style 与占位符不匹配的问题 - 补充 README、配置示例、依赖清单 Co-Authored-By: Claude <noreply@anthropic.com>
142 lines
4.8 KiB
Python
142 lines
4.8 KiB
Python
# -*- 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()),
|
||
})
|