修复db客户端无法推日志的问题

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()

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 剩余日志。"""