qing/blackbird
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev_qing' of https://git.cancerai.top/qing/micro_servers into dev_qing

Browse Source
This commit is contained in:
qing committed 2026-07-27 09:21:20 +08:00
2 parents 179afb2674 08021015b1
commit b790e18024
5 files changed
+196 -86
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

db/example_log_client.py
+11 -12
View File
@@ -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()
dnfs/log_client.py
+11 -12
View File
@@ -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()
smtp/log_client.py
+64 -38
View File
@@ -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")
waf/.gitignore
+1
View File
@@ -1,5 +1,6 @@
# 编译产物
waf.bin
waf
*.exe
# 证书与密钥（不入库）
waf/internal/logclient/logclient.go
+109 -24
View File
@@ -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() {