From 4fe0689cf6ff3a36e86eeea778f4eca781d08a05 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sun, 26 Jul 2026 21:26:17 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Ddb=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=97=A0=E6=B3=95=E6=8E=A8=E6=97=A5=E5=BF=97=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/view_db.py | 85 ++++++++++++++++++++++++++++++++++++++++++++++ dnfs/log_client.py | 43 +++++++++++++++++------ 2 files changed, 118 insertions(+), 10 deletions(-) create mode 100755 db/view_db.py diff --git a/db/view_db.py b/db/view_db.py new file mode 100755 index 0000000..0a3881b --- /dev/null +++ b/db/view_db.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""查看 database.db 各表数据 (非交互式)""" + +import argparse +import sqlite3 +import os +import sys + +DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "database.db") + + +def get_tables(cur): + cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name") + return [r[0] for r in cur.fetchall()] + + +def print_rows(cols, rows): + if not rows: + print("(空)") + return + widths = [len(c) for c in cols] + str_rows = [] + for row in rows: + str_row = [str(v) if v is not None else "NULL" for v in row] + str_rows.append(str_row) + for i, v in enumerate(str_row): + widths[i] = max(widths[i], min(len(v), 50)) + print(" | ".join(c.ljust(widths[i]) for i, c in enumerate(cols))) + print("-+-".join("-" * w for w in widths)) + for str_row in str_rows: + print(" | ".join(v[:50].ljust(widths[i]) for i, v in enumerate(str_row))) + print(f"\n({len(rows)} 行)") + + +def main(): + parser = argparse.ArgumentParser(description="查看 database.db") + parser.add_argument("table", nargs="?", help="要查看的表名 (不指定则列出所有表)") + parser.add_argument("-n", "--limit", type=int, default=50, help="最多显示行数 (默认 50)") + parser.add_argument("-w", "--where", help="WHERE 条件, 如: \"level='ERROR'\"") + parser.add_argument("-s", "--sql", help="自由 SQL 查询 (忽略 table/where/limit)") + args = parser.parse_args() + + if not os.path.exists(DB_PATH): + print(f"数据库不存在: {DB_PATH}", file=sys.stderr) + sys.exit(1) + + conn = sqlite3.connect(DB_PATH) + cur = conn.cursor() + tables = get_tables(cur) + + # 自由 SQL + if args.sql: + cur.execute(args.sql) + cols = [d[0] for d in cur.description] + print_rows(cols, cur.fetchall()) + conn.close() + return + + # 列出所有表 + if not args.table: + print(f"数据库: {DB_PATH}\n") + for t in tables: + cur.execute(f"SELECT COUNT(*) FROM [{t}]") + print(f" {t} ({cur.fetchone()[0]} 行)") + conn.close() + return + + # 查看指定表 + if args.table not in tables: + print(f"表 '{args.table}' 不存在, 可用: {', '.join(tables)}", file=sys.stderr) + sys.exit(1) + + sql = f"SELECT * FROM [{args.table}]" + if args.where: + sql += f" WHERE {args.where}" + sql += f" LIMIT {args.limit}" + + cur.execute(sql) + cols = [d[0] for d in cur.description] + print_rows(cols, cur.fetchall()) + conn.close() + + +if __name__ == "__main__": + main() diff --git a/dnfs/log_client.py b/dnfs/log_client.py index 6ee468c..33b31f4 100644 --- a/dnfs/log_client.py +++ b/dnfs/log_client.py @@ -2,7 +2,7 @@ 统一日志客户端 — 各服务共用。 设计要点： - - 使用 urllib.request（标准库，不引入 requests） + - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush @@ -23,6 +23,7 @@ import threading import queue import time import atexit +from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) from pymsg import MsgClient # noqa: E402 @@ -111,10 +112,11 @@ class LogClient: except queue.Empty: break except queue.Empty: - # 【核心修复 1】：如果没有新日志，必须重置上次刷新时间。 - # 否则下次 timeout 会变成 0，导致疯狂空转。 + # 超时无新日志：如果 batch 中已有积攒的条目，强制 flush + if batch: + self._send(batch) + batch = [] last_flush = time.monotonic() - # 【核心修复 2】：跳出本次循环，避免重复执行下面的 `if should_flush` 判断 continue # 满了或超时了就发 @@ -132,14 +134,35 @@ class LogClient: self._send(batch) def _send(self, batch): - """通过 pymsg 短连接发送一批日志到 dbv2。失败时静默丢弃。""" + """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" + now = datetime.now(timezone.utc).isoformat() try: client = MsgClient(_DBV2_ADDR, _DBV2_PORT, "dnfs-log") - payload = {"_route": "log/ingest", "logs": batch} - client.call(json.dumps(payload), timeout=_TIMEOUT) - except Exception: - # fail-safe：日志服务不可用，静默丢弃，不影响业务 - pass + 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) + except Exception as e: + import sys + print(f"[log_client] 发送失败: {e}", file=sys.stderr) def _shutdown(self): """进程退出时 flush 剩余日志。"""