diff --git a/db/example_log_client.py b/db/example_log_client.py index 022b383..1ffe09a 100644 --- a/db/example_log_client.py +++ b/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 - if batch: - self._send(batch) - batch = [] - last_flush = time.monotonic() - continue + 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: + if should_flush and batch: self._send(batch) batch = [] last_flush = time.monotonic() diff --git a/dnfs/log_client.py b/dnfs/log_client.py index 33b31f4..043d379 100644 --- a/dnfs/log_client.py +++ b/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 - if batch: - self._send(batch) - batch = [] - last_flush = time.monotonic() - continue + 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: + if should_flush and batch: self._send(batch) batch = [] last_flush = time.monotonic() diff --git a/smtp/log_client.py b/smtp/log_client.py index 5bcf8c4..47c1ed7 100644 --- a/smtp/log_client.py +++ b/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.error("dnfs", "数据库连接失败", extra={"errno": 2002}) - logs.push("gateway", "WARNING", "登录失败次数超限", + logs.info("smtp", "验证码发送成功") + logs.error("smtp", "验证码发送失败", extra={"errno": 2002}) + 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】：如果没有新日志，必须重置上次刷新时间。 - # 否则下次 timeout 会变成 0，导致疯狂空转。 - last_flush = time.monotonic() - # 【核心修复 2】：跳出本次循环，避免重复执行下面的 `if should_flush` 判断 - continue + 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: + 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 服务。失败时静默丢弃。""" - data = json.dumps({'logs': batch}).encode('utf-8') - req = urllib.request.Request( - self._url, data=data, - headers={'Content-Type': 'application/json'}, - method='POST' - ) + """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" + now = datetime.now(timezone.utc).isoformat() try: - with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: - resp.read() # 消费响应体 - except (urllib.error.URLError, urllib.error.HTTPError, OSError): - # fail-safe：日志服务不可用，静默丢弃，不影响业务 - pass - except Exception: - pass + 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) + 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") diff --git a/waf/.gitignore b/waf/.gitignore index 43fdf7f..d822e13 100644 --- a/waf/.gitignore +++ b/waf/.gitignore @@ -1,5 +1,6 @@ # 编译产物 waf.bin +waf *.exe # 证书与密钥（不入库） diff --git a/waf/internal/logclient/logclient.go b/waf/internal/logclient/logclient.go index 2dd597d..3dcb844 100644 --- a/waf/internal/logclient/logclient.go +++ b/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" - batchSize = 50 - flushInterval = 2 * time.Second - requestTimeout = 5 * time.Second + dbAddr = "127.0.0.1:8005" + batchSize = 50 + flushInterval = 2 * 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), - client: &http.Client{Timeout: requestTimeout}, - stop: make(chan struct{}), + ch: make(chan Entry, 256), + 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} - data, err := json.Marshal(payload) + now := time.Now().UTC().Format(time.RFC3339) + 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() {