From bdbea6ac400bb486d5ffdaea1f87b969a808bdb9 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 03:26:19 +0800 Subject: [PATCH] fix(log_client): update example_log_client template to use RpcClient - Replace legacy MsgClient (low-level TCP frames) with RpcClient (high-level RPC) to match dbv2's RpcServer interface; the old MsgClient-based version was incompatible with the current RPC routing protocol - Apply all hardening fixes from the service copies: - Outer try/except in _flush_loop to prevent any exception from killing the background thread - _safe_json() helper for unserializable extra fields (fallback to str/repr) - Timestamp generated at push() (enqueue time) instead of _send() time - Event-based flush() synchronization instead of unreliable sleep(0.6) - Queue maxsize=10000 with drop-oldest policy to prevent unbounded memory - put_nowait with Full handling ensures push() never blocks business threads - RpcClient construction wrapped in try/except - Add explicit service_name parameter example for all 4 services in comments - This is the template file referenced by db/README.md and copied by new services; keeping it in sync prevents propagating old bugs --- db/example_log_client.py | 199 ++++++++++++++++++++++++++------------- 1 file changed, 135 insertions(+), 64 deletions(-) diff --git a/db/example_log_client.py b/db/example_log_client.py index 1ffe09a..8a41e67 100644 --- a/db/example_log_client.py +++ b/db/example_log_client.py @@ -1,13 +1,14 @@ """ -统一日志客户端 SDK — 各服务复制即用。 +统一日志客户端 SDK — 各服务复制即用（模板）。 设计要点： - - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 + - 通过 pymsg RpcClient（高层 RPC 封装）调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + - 时间戳：日志入队时即记录 ts，反映事件真实发生时间 -复制到目标服务目录后，按需修改 service_name 即可：: +复制到目标服务目录后，按需修改底部全局单例的 service_name 即可： from log_client import logs @@ -27,22 +28,36 @@ import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import MsgClient # noqa: E402 +from pymsg import RpcClient, RpcError # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── -_DBV2_ADDR = '127.0.0.1' -_DBV2_PORT = 8005 +_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) _BATCH_SIZE = 50 # 队列满 50 条时 flush _FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush _TIMEOUT = 5 # socket 超时（秒） +_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目 + + +def _safe_json(obj): + """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。""" + try: + return json.dumps(obj, ensure_ascii=False, default=str) + except Exception: + return repr(obj) + + +def _now_iso(): + """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。""" + return datetime.now(timezone.utc).isoformat() class LogClient: """异步批量日志推送客户端。 - 通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。 + 通过 RpcClient（pymsg 高层 RPC 封装）逐条 INSERT 到 dbv2 的 Logs 表。 service_name 用于在 dbv2 连接中标识来源，各服务复制时自行修改。 """ @@ -52,9 +67,10 @@ class LogClient: self._service_name = service_name self._batch_size = batch_size self._flush_interval = flush_interval - self._queue = queue.Queue() + self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) self._stop = threading.Event() self._flush_signal = threading.Event() + self._flush_done = threading.Event() self._thread = threading.Thread(target=self._flush_loop, daemon=True, name='log-client') self._thread.start() @@ -63,14 +79,30 @@ class LogClient: # ── 公开接口 ──────────────────────────────────────────────────────── def push(self, service, level, message, trace_id=None, extra=None): - """推送一条日志（非阻塞，立即返回）。""" - self._queue.put({ + """推送一条日志（非阻塞，立即返回）。 + + 队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。 + """ + entry = { + 'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间 'service': service, 'level': level, 'message': message, 'trace_id': trace_id, 'extra': extra, - }) + } + try: + self._queue.put_nowait(entry) + except queue.Full: + # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略） + try: + self._queue.get_nowait() + except queue.Empty: + pass + try: + self._queue.put_nowait(entry) + except queue.Full: + pass # 极端竞争下仍失败则静默丢弃 def debug(self, service, message, **kw): self.push(service, 'DEBUG', message, **kw) @@ -89,8 +121,15 @@ class LogClient: def flush(self): """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" + self._flush_done.clear() self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - time.sleep(0.6) # 等待后台线程响应 + # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量） + self._flush_done.wait(timeout=self._flush_interval + 2.0) + # 主线程排空队列中剩余条目 + self._drain_queue_to_db() + + def _drain_queue_to_db(self): + """排空主线程可见的队列中剩余条目，直接发送到 db。""" batch = [] while True: try: @@ -102,73 +141,103 @@ class LogClient: break if batch: self._send(batch) - self._flush_signal.clear() # ── 内部实现 ──────────────────────────────────────────────────────── def _flush_loop(self): - """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。""" + """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。 + + 外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等） + 都不会死亡，避免日志队列无限增长导致 OOM。 + """ batch = [] last_flush = time.monotonic() while not self._stop.is_set(): - timeout = self._flush_interval - (time.monotonic() - last_flush) - chk_timeout = max(0.0, min(timeout, 0.5)) try: - item = self._queue.get(timeout=chk_timeout) - batch.append(item) - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass + timeout = self._flush_interval - (time.monotonic() - last_flush) + chk_timeout = max(0.1, min(timeout, 0.5)) + try: + item = self._queue.get(timeout=chk_timeout) + batch.append(item) + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + pass - should_flush = ( - len(batch) >= self._batch_size or - self._flush_signal.is_set() or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush and batch: - self._send(batch) - batch = [] - last_flush = time.monotonic() + should_flush = ( + len(batch) >= self._batch_size or + self._flush_signal.is_set() or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush and batch: + try: + self._send(batch) + except Exception as e: + # _send 内部已有细粒度异常处理，此处是最终安全网 + print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}", + file=_sys.stderr) + batch = [] + last_flush = time.monotonic() + if self._flush_signal.is_set(): + self._flush_signal.clear() + self._flush_done.set() + except Exception as e: + # 最外层安全网：任何意外异常都不应杀死日志线程 + print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}", + file=_sys.stderr) + time.sleep(0.5) # 防止异常风暴导致 CPU 满载 # 退出前 flush 剩余 if batch: - self._send(batch) + try: + self._send(batch) + except Exception as e: + print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr) + self._flush_done.set() def _send(self, batch): - """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" - now = datetime.now(timezone.utc).isoformat() + """通过 pymsg RpcClient（execute 路由）批量发送到 dbv2 Logs 表。 + + 整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe）， + 不抛异常给调用方。 + """ try: - client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - for entry in batch: - extra_val = entry.get("extra") - payload = { - "_route": "execute", - "action": "insert", - "table": "Logs", - "values": { - "ts": now, - "service": entry.get("service"), - "level": entry.get("level"), - "message": entry.get("message"), - "trace_id": entry.get("trace_id"), - "extra": json.dumps(extra_val, ensure_ascii=False) - if extra_val is not None else None, - }, - } - resp = client.call(json.dumps(payload), timeout=_TIMEOUT) - resp_obj = json.loads(resp) - if not resp_obj.get("success"): - import sys - print(f"[log_client] INSERT 失败: {resp_obj.get('error')}", - file=sys.stderr) + client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") except Exception as e: - import sys - print(f"[log_client] 发送失败: {e}", file=sys.stderr) + print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr) + return + + for entry in batch: + extra_val = entry.get("extra") + try: + extra_json = _safe_json(extra_val) if extra_val is not None else None + except Exception: + extra_json = None + payload = { + "action": "insert", + "table": "Logs", + "values": { + "ts": entry.get("ts", _now_iso()), + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": extra_json, + }, + } + try: + client.call("execute", payload, timeout=_TIMEOUT) + except RpcError as e: + # 业务错误（如表结构不符）：打印后继续下一条 + print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) + except Exception as e: + # 网络错误：db 不可用，放弃本批剩余（fail-safe） + print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + break def _shutdown(self): """进程退出时 flush 剩余日志。""" @@ -180,5 +249,7 @@ class LogClient: # 各服务复制时，把 "example" 改为自己的服务名即可。 # 例如 dnfs/log_client.py → logs = LogClient(service_name="dnfs") # 例如 smtp/log_client.py → logs = LogClient(service_name="smtp") +# 例如 static/log_client.py → logs = LogClient(service_name="static") +# 例如 confs/log_client.py → logs = LogClient(service_name="confs") -logs = LogClient() +logs = LogClient(service_name="example")