更新msg客户端

db/example_log_client.py

@@ -54,6 +54,7 @@ class LogClient:
         self._flush_interval = flush_interval
         self._queue = queue.Queue()
         self._stop = threading.Event()
+        self._flush_signal = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -87,7 +88,9 @@ class LogClient:
         self.push(service, 'CRITICAL', message, **kw)
 
     def flush(self):
-        """手动 flush 队列中剩余的日志。"""
+        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
+        time.sleep(0.6)  # 等待后台线程响应
         batch = []
         while True:
             try:
@@ -99,6 +102,7 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
+        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
@@ -108,30 +112,25 @@ class LogClient:
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = max(0, self._flush_interval - (time.monotonic() - last_flush))
+            timeout = self._flush_interval - (time.monotonic() - last_flush)
+            chk_timeout = max(0.0, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=timeout)
+                item = self._queue.get(timeout=chk_timeout)
                 batch.append(item)
-                # 继续非阻塞地取，直到满 batch_size
                 while len(batch) < self._batch_size:
                     try:
                         batch.append(self._queue.get_nowait())
                     except queue.Empty:
                         break
             except queue.Empty:
-                # 超时无新日志：如果 batch 中已有积攒的条目，强制 flush
+                pass
-                if batch:
-                    self._send(batch)
-                    batch = []
-                last_flush = time.monotonic()
-                continue
 
-            # 满了或超时了就发
             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:
+            if should_flush and batch:
                 self._send(batch)
                 batch = []
                 last_flush = time.monotonic()

dnfs/log_client.py

@@ -47,6 +47,7 @@ class LogClient:
         self._flush_interval = flush_interval
         self._queue = queue.Queue()
         self._stop = threading.Event()
+        self._flush_signal = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -80,7 +81,9 @@ class LogClient:
         self.push(service, 'CRITICAL', message, **kw)
 
     def flush(self):
-        """手动 flush 队列中剩余的日志。"""
+        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
+        time.sleep(0.6)  # 等待后台线程响应
         batch = []
         while True:
             try:
@@ -92,6 +95,7 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
+        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
@@ -101,30 +105,25 @@ class LogClient:
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = max(0, self._flush_interval - (time.monotonic() - last_flush))
+            timeout = self._flush_interval - (time.monotonic() - last_flush)
+            chk_timeout = max(0.0, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=timeout)
+                item = self._queue.get(timeout=chk_timeout)
                 batch.append(item)
-                # 继续非阻塞地取，直到满 batch_size
                 while len(batch) < self._batch_size:
                     try:
                         batch.append(self._queue.get_nowait())
                     except queue.Empty:
                         break
             except queue.Empty:
-                # 超时无新日志：如果 batch 中已有积攒的条目，强制 flush
+                pass
-                if batch:
-                    self._send(batch)
-                    batch = []
-                last_flush = time.monotonic()
-                continue
 
-            # 满了或超时了就发
             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:
+            if should_flush and batch:
                 self._send(batch)
                 batch = []
                 last_flush = time.monotonic()

smtp/log_client.py

@@ -1,8 +1,8 @@
 """
-统一日志客户端 — 各服务共用。
+统一日志客户端 — 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。
 
 设计要点：
-  - 使用 urllib.request（标准库，不引入 requests）
+  - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
   - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
   - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
@@ -10,39 +10,49 @@
 用法：
     from log_client import logs
 
-    logs.info("dnfs", "节点新增成功 id=42")
+    logs.info("smtp", "验证码发送成功")
-    logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
+    logs.error("smtp", "验证码发送失败", extra={"errno": 2002})
-    logs.push("gateway", "WARNING", "登录失败次数超限",
+    logs.push("smtp", "WARNING", "登录失败次数超限",
               trace_id="req-abc", extra={"ip": "1.2.3.4"})
 """
 
 import json
+import os as _os
+import sys as _sys
 import threading
 import queue
 import time
-import urllib.request
-import urllib.error
 import atexit
+from datetime import datetime, timezone
+
+_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
+from pymsg import MsgClient  # noqa: E402
 
 # ── 配置 ──────────────────────────────────────────────────────────────
 
-DB_LOG_URL = 'http://127.0.0.1:8005/log/ingest'
+_DBV2_ADDR = '127.0.0.1'
+_DBV2_PORT = 8005
 
 _BATCH_SIZE = 50         # 队列满 50 条时 flush
 _FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
-_TIMEOUT = 5             # HTTP 请求超时（秒）
+_TIMEOUT = 5             # socket 超时（秒）
 
 
 class LogClient:
-    """异步批量日志推送客户端。"""
+    """异步批量日志推送客户端。
 
-    def __init__(self, url=DB_LOG_URL, batch_size=_BATCH_SIZE,
+    通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。
+    """
+
+    def __init__(self, service_name="smtp",
+                 batch_size=_BATCH_SIZE,
                  flush_interval=_FLUSH_INTERVAL):
-        self._url = url
+        self._service_name = service_name
         self._batch_size = batch_size
         self._flush_interval = flush_interval
         self._queue = queue.Queue()
         self._stop = threading.Event()
+        self._flush_signal = threading.Event()
         self._thread = threading.Thread(target=self._flush_loop,
                                         daemon=True, name='log-client')
         self._thread.start()
@@ -76,7 +86,11 @@ class LogClient:
         self.push(service, 'CRITICAL', message, **kw)
 
     def flush(self):
-        """手动 flush 队列中剩余的日志。"""
+        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
+        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
+        # 等待后台线程响应（queue.get 超时最长 0.5s）
+        time.sleep(0.6)
+        # 排空主队列中尚未被线程取走的条目
         batch = []
         while True:
             try:
@@ -88,6 +102,7 @@ class LogClient:
                 break
         if batch:
             self._send(batch)
+        self._flush_signal.clear()
 
     # ── 内部实现 ────────────────────────────────────────────────────────
 
@@ -97,29 +112,26 @@ class LogClient:
         last_flush = time.monotonic()
 
         while not self._stop.is_set():
-            timeout = max(0, self._flush_interval - (time.monotonic() - last_flush))
+            timeout = self._flush_interval - (time.monotonic() - last_flush)
+            # 限制最长阻塞时间，以便及时响应 flush() 信号
+            chk_timeout = max(0.0, min(timeout, 0.5))
             try:
-                item = self._queue.get(timeout=timeout)
+                item = self._queue.get(timeout=chk_timeout)
                 batch.append(item)
-                # 继续非阻塞地取，直到满 batch_size
                 while len(batch) < self._batch_size:
                     try:
                         batch.append(self._queue.get_nowait())
                     except queue.Empty:
                         break
             except queue.Empty:
-                # 【核心修复 1】：如果没有新日志，必须重置上次刷新时间。
+                pass
-                # 否则下次 timeout 会变成 0，导致疯狂空转。
-                last_flush = time.monotonic()
-                # 【核心修复 2】：跳出本次循环，避免重复执行下面的 `if should_flush` 判断
-                continue
 
-            # 满了或超时了就发
             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:
+            if should_flush and batch:
                 self._send(batch)
                 batch = []
                 last_flush = time.monotonic()
@@ -129,21 +141,35 @@ class LogClient:
             self._send(batch)
 
     def _send(self, batch):
-        """发送一批日志到 db 服务。失败时静默丢弃。"""
+        """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。"""
-        data = json.dumps({'logs': batch}).encode('utf-8')
+        now = datetime.now(timezone.utc).isoformat()
-        req = urllib.request.Request(
-            self._url, data=data,
-            headers={'Content-Type': 'application/json'},
-            method='POST'
-        )
         try:
-            with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
+            client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
-                resp.read()  # 消费响应体
+            for entry in batch:
-        except (urllib.error.URLError, urllib.error.HTTPError, OSError):
+                extra_val = entry.get("extra")
-            # fail-safe：日志服务不可用，静默丢弃，不影响业务
+                payload = {
-            pass
+                    "_route": "execute",
-        except Exception:
+                    "action": "insert",
-            pass
+                    "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 剩余日志。"""
@@ -153,4 +179,4 @@ class LogClient:
 
 # ── 全局单例 ──────────────────────────────────────────────────────────
 
-logs = LogClient()
+logs = LogClient(service_name="smtp")

waf/.gitignore

@@ -1,5 +1,6 @@
 # 编译产物
 waf.bin
+waf
 *.exe
 
 # 证书与密钥（不入库）

waf/internal/logclient/logclient.go

@@ -1,24 +1,30 @@
-// Package logclient 提供异步批量日志推送客户端，将日志发送到 db 服务的 /log/ingest 接口。
+// Package logclient 提供异步批量日志推送客户端。
 //
-// 设计要点（与 Python log_client.py 对齐）：
+// 通过 pymsg TCP 协议调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。
+//
+// 设计要点（与 Python example_log_client.py 对齐）：
 //   - 异步非阻塞：内存 channel + 后台 goroutine 批量推送，不拖慢业务
 //   - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
 //   - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
+//
+// 协议：pymsg 长度前缀帧（4 字节 big-endian uint32 长度 + UTF-8 载荷）
 package logclient
 
 import (
-	"bytes"
+	"encoding/binary"
 	"encoding/json"
-	"net/http"
+	"fmt"
+	"io"
+	"net"
 	"sync"
 	"time"
 )
 
 const (
-	defaultURL     = "http://127.0.0.1:8005/log/ingest"
+	dbAddr        = "127.0.0.1:8005"
-	batchSize      = 50
+	batchSize     = 50
-	flushInterval  = 2 * time.Second
+	flushInterval = 2 * time.Second
-	requestTimeout = 5 * time.Second
+	socketTimeout = 5 * time.Second
 )
 
 // Entry 单条日志。
@@ -30,11 +36,23 @@ type Entry struct {
 	Extra   map[string]any `json:"extra,omitempty"`
 }
 
+// dbV2Request execute 路由的 INSERT 请求体。
+type dbV2Request struct {
+	Route  string         `json:"_route"`
+	Action string         `json:"action"`
+	Table  string         `json:"table"`
+	Values map[string]any `json:"values"`
+}
+
+// dbV2Response execute 路由的通用响应。
+type dbV2Response struct {
+	Success bool   `json:"success"`
+	Error   string `json:"error,omitempty"`
+}
+
 // Client 异步批量日志推送客户端。
 type Client struct {
-	url     string
 	ch      chan Entry
-	client  *http.Client
 	stop    chan struct{}
 	wg      sync.WaitGroup
 	stopped bool
@@ -46,16 +64,14 @@ var global *Client
 
 // Init 初始化全局日志客户端并启动后台 flush goroutine。
 func Init() {
-	global = New(defaultURL)
+	global = New()
 }
 
 // New 创建客户端实例。
-func New(url string) *Client {
+func New() *Client {
 	c := &Client{
-		url:    url,
+		ch:   make(chan Entry, 256),
-		ch:     make(chan Entry, 256),
+		stop: make(chan struct{}),
-		client: &http.Client{Timeout: requestTimeout},
-		stop:   make(chan struct{}),
 	}
 	c.wg.Add(1)
 	go c.flushLoop()
@@ -153,21 +169,90 @@ func (c *Client) flushLoop() {
 }
 
 func (c *Client) send(batch []Entry) {
-	payload := map[string]any{"logs": batch}
+	now := time.Now().UTC().Format(time.RFC3339)
-	data, err := json.Marshal(payload)
+	for _, entry := range batch {
+		c.sendOne(entry, now)
+	}
+}
+
+// sendOne 通过 pymsg TCP 协议发送单条日志到 dbv2。
+func (c *Client) sendOne(entry Entry, ts string) {
+	extraJSON := ""
+	if entry.Extra != nil {
+		b, err := json.Marshal(entry.Extra)
+		if err != nil {
+			return
+		}
+		extraJSON = string(b)
+	}
+
+	values := map[string]any{
+		"ts":      ts,
+		"service": entry.Service,
+		"level":   entry.Level,
+		"message": entry.Message,
+	}
+	if entry.TraceID != "" {
+		values["trace_id"] = entry.TraceID
+	}
+	if extraJSON != "" {
+		values["extra"] = extraJSON
+	}
+
+	req := dbV2Request{
+		Route:  "execute",
+		Action: "insert",
+		Table:  "Logs",
+		Values: values,
+	}
+	data, err := json.Marshal(req)
 	if err != nil {
 		return
 	}
-	req, err := http.NewRequest("POST", c.url, bytes.NewReader(data))
+
+	// pymsg 协议：4 字节 big-endian 长度 + UTF-8 载荷
+	resp, err := pymsgCall(data)
 	if err != nil {
-		return
+		return // fail-safe
 	}
-	req.Header.Set("Content-Type", "application/json")
+
-	resp, err := c.client.Do(req)
+	var r dbV2Response
+	if json.Unmarshal(resp, &r) != nil || !r.Success {
+		fmt.Printf("[logclient] INSERT failed: %s\n", r.Error)
+	}
+}
+
+// pymsgCall 发送一条 pymsg 请求并读取响应（短连接模式）。
+func pymsgCall(payload []byte) ([]byte, error) {
+	conn, err := net.DialTimeout("tcp", dbAddr, socketTimeout)
 	if err != nil {
-		return // fail-safe：静默丢弃
+		return nil, err
 	}
-	resp.Body.Close()
+	defer conn.Close()
+
+	conn.SetDeadline(time.Now().Add(socketTimeout))
+
+	// 发送：4 字节长度前缀 + 载荷
+	header := make([]byte, 4)
+	binary.BigEndian.PutUint32(header, uint32(len(payload)))
+	if _, err := conn.Write(header); err != nil {
+		return nil, err
+	}
+	if _, err := conn.Write(payload); err != nil {
+		return nil, err
+	}
+
+	// 接收：4 字节长度前缀 + 载荷
+	if _, err := io.ReadFull(conn, header); err != nil {
+		return nil, err
+	}
+	respLen := binary.BigEndian.Uint32(header)
+	resp := make([]byte, respLen)
+	if _, err := io.ReadFull(conn, resp); err != nil {
+		return nil, err
+	}
+
+	return resp, nil
 }
 
 func (c *Client) flush() {