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

feat(liblogc): 统一集中式日志客户端库 + 全量迁移

Browse Source 
新增:
- liblogc/: C++ 统一日志库 (libus_logc.so)，支持 set_endpoint/set_service_name/set_level
- pymsg/pymsg/logc.py: Python 统一日志模块，基于 AsyncRpcQueue

C++ 迁移 (删除 6 份本地 LogC.hpp 副本):
- ca, gateway: <us/LogC.hpp> + set_service_name
- captcha2, confs2, smtp3: using qing::LogC + set_service_name
- core: ConfsLogC → qing::LogC (ConfsTask.cpp)
- 所有 Makefile 添加 -lus_logc

Python 迁移 (6 个 log_client.py → 4 行薄包装):
- smtp, redirect80, confs, captcha, dnfs, static
- 全部委托到 pymsg.logc.LogClient

Go (waf): 保持现状，自成体系
This commit is contained in:
qing committed 2026-08-10 04:51:43 +00:00
1 parent 01ae877673
commit aa3741b861
32 files changed
+608 -2129
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ca/Makefile
+2 -2
View File
@@ -3,7 +3,7 @@ APP_NAME := ca
INSTL_DIR := /usr/local/bin
CPP := g++
INCS := -I inc/ -I ../libmsg/inc
MY_LIBS := -L/usr/lib -L../libmsg/build -lus_http -lus_httpclient -lus_crypt -lus_msg
MY_LIBS := -L/usr/lib -L../libmsg/build -lus_http -lus_httpclient -lus_crypt -lus_msg -lus_logc
LIBS := -pthread -lcurl -lssl -lz -lcrypto -lboost_json -lboost_program_options -Wl,-rpath,$(shell cd ../libmsg/build && pwd)
CPPFLAGS := --std=c++20 -pipe -pthread # -Wall -Werror
BUILD_DIR = build
@@ -13,7 +13,7 @@ $(shell mkdir -p $(BUILD_DIR))
all: $(BUILD_DIR)/$(APP_NAME).out
$(BUILD_DIR)/$(APP_NAME).out: $(BUILD_DIR)/main.o
$(CPP) $^ $(LIBS) $(MY_LIBS) -o $@
$(BUILD_DIR)/main.o: src/main.cpp inc/DbC.hpp inc/LogC.hpp inc/Conf.hpp inc/ExpBase.hpp inc/LocalTime.hpp
$(BUILD_DIR)/main.o: src/main.cpp inc/DbC.hpp inc/Conf.hpp inc/ExpBase.hpp inc/LocalTime.hpp
$(CPP) $(CPPFLAGS) -c $< $(INCS) -o $@
clear:
@rm -rvf build/
ca/inc/LogC.hpp
-235
View File
@@ -1,235 +0,0 @@
#pragma once
/** LogC — ca 统一日志客户端（C++ 版，对应各 Python 服务的 log_client.py）
*
* 设计要点（与 Python 版对齐）：
* - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
* - 异步非阻塞：基于 pmc::net::AsyncQueue 内存队列 + 后台线程批量推送，不拖慢业务
* - fail-safe：db 服务不可用时静默丢弃（仅 stderr 提示），绝不影响业务
* - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
* - 背压保护：队列上限 10000 条，满时丢最旧日志，防止内存无限增长
*
* 用法：
* #include "LogC.hpp"
*
* qing::LogC::instance().info("ca", "启动完成");
* qing::LogC::instance().error("ca", "数据库连接失败",
* std::nullopt, R"({"errno":2002})");
*
* // 进程退出前：
* qing::LogC::instance().shutdown(); // 停止后台线程并 flush 剩余日志（幂等）
*
* // 设置日志级别（main 启动时调用，低于此级别的日志直接丢弃）
* qing::LogC::instance().set_level("warn"); // 只输出 warn/error
*/
#include <atomic>
#include <chrono>
#include <cstdio>
#include <ctime>
#include <mutex>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
#include <boost/json.hpp>
#include <us/AsyncQueue.hpp>
#include <us/Msg.hpp>
namespace qing {
namespace json = boost::json;
class LogC {
public:
struct Entry {
std::string service;
std::string level;
std::string message;
std::optional<std::string> trace_id;
std::optional<std::string> extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT）
};
// 日志级别 → 数值（大写兼容）
static const std::unordered_map<std::string, int>& level_order() {
static const std::unordered_map<std::string, int> k = {
{"debug", 0}, {"info", 1}, {"warn", 2}, {"warning", 2},
{"error", 3}, {"critical", 3},
{"DEBUG", 0}, {"INFO", 1}, {"WARN", 2}, {"WARNING", 2},
{"ERROR", 3}, {"CRITICAL", 3},
};
return k;
}
static LogC& instance() {
static LogC inst;
return inst;
}
// ---- 公开接口 --------------------------------------------------------
// 设置日志级别阈值（debug|info|warn|error，大小写不限）
// 低于此级别的日志在 push 时直接丢弃
void set_level(const std::string& level) {
auto it = level_order().find(level);
if (it != level_order().end()) {
_level_threshold.store(it->second, std::memory_order_relaxed);
}
}
// 推送一条日志（非阻塞，立即返回）
void push(const std::string& service, const std::string& level,
const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
// 日志级别过滤：低于阈值直接丢弃
auto it = level_order().find(level);
int lvl = (it != level_order().end()) ? it->second : 1;
if (lvl < _level_threshold.load(std::memory_order_relaxed)) return;
_queue.push(Entry{service, level, message,
std::move(trace_id), std::move(extra)});
}
void debug(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "DEBUG", message, std::move(trace_id), std::move(extra));
}
void info(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "INFO", message, std::move(trace_id), std::move(extra));
}
void warning(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "WARNING", message, std::move(trace_id), std::move(extra));
}
void error(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "ERROR", message, std::move(trace_id), std::move(extra));
}
void critical(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "CRITICAL", message, std::move(trace_id), std::move(extra));
}
// 手动 flush：通知后台线程立即发送当前积攒的批次，并等待其完成
void flush() {
_queue.flush();
}
// 停止后台线程并 flush 剩余日志（进程退出前调用，幂等）
void shutdown() {
_queue.shutdown();
}
// ---- 配置（构造后、首次推送前可改）------------------------------------
void set_endpoint(const std::string& addr, int port) {
std::lock_guard<std::mutex> lock(_endpoint_mtx);
_db_addr = addr;
_db_port = port;
}
private:
LogC()
: _queue(
[this](std::vector<Entry>&& batch) { this->_send_batch(std::move(batch)); },
{_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5},
[](const std::exception& e, std::vector<Entry>&&) {
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
})
{
_queue.start();
}
~LogC() { shutdown(); }
LogC(const LogC&) = delete;
LogC& operator=(const LogC&) = delete;
// 通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表
void _send_batch(const std::vector<Entry>& batch) {
std::string addr;
int port;
{
std::lock_guard<std::mutex> lock(_endpoint_mtx);
addr = _db_addr;
port = _db_port;
}
const std::string now = _utc_now_iso(); // 批次共享同一时间戳（对齐 Python 版）
try {
pmc::net::MsgClient client(addr, port, "ca-log");
for (const auto& e : batch) {
json::object values;
values["ts"] = now;
values["service"] = e.service;
values["level"] = e.level;
values["message"] = e.message;
values["trace_id"] = e.trace_id ? json::value(*e.trace_id)
: json::value(nullptr);
values["extra"] = e.extra ? json::value(*e.extra)
: json::value(nullptr);
json::object payload;
payload["_route"] = "execute";
payload["action"] = "insert";
payload["table"] = "Logs";
payload["values"] = std::move(values);
const std::string resp = client.call(
json::serialize(payload), _TIMEOUT,
/*retry_on_refused=*/false); // 日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞重试
auto obj = json::parse(resp).as_object();
if (!obj.at("success").as_bool()) {
std::string err = "unknown";
if (auto* ev = obj.if_contains("error")) {
if (ev->is_string()) err = ev->as_string().c_str();
}
std::cerr << "[LogC] INSERT 失败: " << err << std::endl;
}
}
} catch (const std::exception& e) {
// fail-safe：db 不可用时静默丢弃，绝不影响业务
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
}
}
// UTC ISO8601 时间戳（对齐 Python datetime.now(timezone.utc).isoformat()）
// 形如 2026-07-29T08:30:00.123456+00:00，字典序即时间序，便于 db 按 ts 清理
static std::string _utc_now_iso() {
const auto now = std::chrono::system_clock::now();
const auto t = std::chrono::system_clock::to_time_t(now);
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
now.time_since_epoch()).count() % 1000000;
std::tm tm{};
::gmtime_r(&t, &tm);
char buf[40];
std::snprintf(buf, sizeof buf,
"%04d-%02d-%02dT%02d:%02d:%02d.%06d+00:00",
tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
tm.tm_hour, tm.tm_min, tm.tm_sec, static_cast<int>(us));
return buf;
}
// ---- 配置项 -----------------------------------------------------------
std::string _db_addr = "127.0.0.1";
int _db_port = 8005;
std::mutex _endpoint_mtx; // 保护 _db_addr/_db_port 读写（set_endpoint 与后台线程并发）
std::atomic<int> _level_threshold{1}; // 日志级别阈值，默认 INFO(1)
static constexpr size_t _BATCH_SIZE = 50; // 队列满 50 条时 flush
static constexpr double _FLUSH_INTERVAL = 2.0; // 超时 2 秒强制 flush
static constexpr size_t _MAX_QUEUE_SIZE = 10000; // 队列上限，满时丢最旧
static constexpr double _TIMEOUT = 5.0; // socket 超时（秒）
// ---- 核心队列 ---------------------------------------------------------
pmc::net::AsyncQueue<Entry> _queue;
};
} // namespace qing
ca/src/main.cpp
+2 -1
View File
@@ -25,7 +25,7 @@
#include "LocalTime.hpp"
#include "Conf.hpp"
#include "DbC.hpp"
#include "LogC.hpp"
#include <us/LogC.hpp>
using namespace pmc::net;
using namespace qing;
@@ -971,6 +971,7 @@ int main(int argc, char **argv) {
/* 初始化下游客户端地址 */
DbC::init(_DB_HOST, _DB_PORT);
logs().set_endpoint(_DB_HOST, _DB_PORT);
logs().set_service_name("ca-log");
/* 设置日志级别过滤 */
logs().set_level(_LOG_LEVEL);
captcha/log_client.py
+2 -156
View File
@@ -1,158 +1,4 @@
"""
统一日志客户端 — captcha 服务副本（基于 pymsg.AsyncRpcQueue）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
用法（与旧版完全兼容）：
from log_client import logs
logs.info("captcha", "验证码已生成 token=abc123")
logs.error("captcha", "图片生成失败", extra={"errno": 2002})
logs.push("captcha", "WARNING", "Token 校验失败",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import os as _os
import sys as _sys
import atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
class LogClient:
"""异步批量日志推送客户端（API 与旧版完全兼容）。"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(),
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
except Exception as e:
print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
break
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
# ── 全局单例 ──────────────────────────────────────────────────────────
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
logs = LogClient(service_name="captcha")
captcha2/Makefile
+1 -1
View File
@@ -1,7 +1,7 @@
CXX := g++
CXXFLAGS := -std=c++20 -Wall -Wextra -O2
INCS := -I./inc -I/usr/include/freetype2
LIBS := -lus_msg -lboost_json -lboost_program_options -lfreetype -lpthread -lm
LIBS := -lus_msg -lus_logc -lboost_json -lboost_program_options -lfreetype -lpthread -lm
TARGET := captcha2.bin
SRC := src/main.cpp
captcha2/inc/LogC.hpp
-139
View File
@@ -1,139 +0,0 @@
/*
* LogC.hpp — captcha2 异步日志客户端（简化版）
*
* 基于 libmsg AsyncQueue，异步批量推送日志到 db:8005 的 Logs 表。
* - 内存队列 + 后台线程攒批（满 50 条或 2 秒 flush）
* - fail-safe：db 不可用时静默丢弃
* - API: info() / warning() / error()
*/
#pragma once
#include <us/Msg.hpp>
#include <us/AsyncQueue.hpp>
#include <boost/json.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <cstdlib>
#include <atomic>
#include <unordered_map>
namespace captcha2 {
struct LogEntry {
std::string ts;
std::string service;
std::string level;
std::string message;
std::string trace_id;
std::string extra;
};
class LogC {
public:
// 日志级别 → 数值（大写兼容）
static const std::unordered_map<std::string, int>& level_order() {
static const std::unordered_map<std::string, int> k = {
{"debug", 0}, {"info", 1}, {"warn", 2}, {"warning", 2},
{"error", 3}, {"critical", 3},
{"DEBUG", 0}, {"INFO", 1}, {"WARN", 2}, {"WARNING", 2},
{"ERROR", 3}, {"CRITICAL", 3},
};
return k;
}
static LogC& instance() {
static LogC inst;
return inst;
}
// 设置日志级别阈值（debug|info|warn|error，大小写不限）
// 低于此级别的日志在 push 时直接丢弃
void set_level(const std::string& level) {
auto it = level_order().find(level);
if (it != level_order().end()) {
level_threshold_.store(it->second, std::memory_order_relaxed);
}
}
void info(const std::string& svc, const std::string& msg) {
push(svc, "INFO", msg);
}
void warning(const std::string& svc, const std::string& msg) {
push(svc, "WARNING", msg);
}
void error(const std::string& svc, const std::string& msg) {
push(svc, "ERROR", msg);
}
private:
LogC() {
const char* host = std::getenv("DB_HOST");
const char* port = std::getenv("DB_PORT");
db_host_ = host ? host : "127.0.0.1";
db_port_ = port ? std::atoi(port) : 8005;
queue_ = std::make_unique<pmc::net::AsyncQueue<LogEntry>>(
[this](std::vector<LogEntry>&& batch) { send_batch(std::move(batch)); },
pmc::net::AsyncQueueConfig{.batch_size = 50, .flush_interval = 2.0, .max_queue_size = 10000},
[](const std::exception& e, std::vector<LogEntry>&&) {
std::cerr << "[captcha2/log] send failed: " << e.what() << std::endl;
});
queue_->start();
}
~LogC() { if (queue_) queue_->shutdown(); }
void push(const std::string& svc, const std::string& level, const std::string& msg) {
// 日志级别过滤：低于阈值直接丢弃
auto it = level_order().find(level);
int lvl = (it != level_order().end()) ? it->second : 1;
if (lvl < level_threshold_.load(std::memory_order_relaxed)) return;
auto now = std::chrono::system_clock::now();
auto t = std::chrono::system_clock::to_time_t(now);
char buf[32];
std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", std::gmtime(&t));
LogEntry e;
e.ts = buf;
e.service = svc;
e.level = level;
e.message = msg;
queue_->push(std::move(e));
}
void send_batch(std::vector<LogEntry>&& batch) {
try {
// P3-4: 复用单连接发送整批日志，超时 1.0s（日志非关键路径，db2 慢时不阻塞 flush 线程过久）
pmc::net::MsgClient client(db_host_, db_port_, "captcha2-log");
for (auto& e : batch) {
boost::json::object vals;
vals["ts"] = e.ts;
vals["service"] = e.service;
vals["level"] = e.level;
vals["message"] = e.message;
vals["trace_id"] = e.trace_id.empty() ? nullptr : boost::json::value(e.trace_id);
vals["extra"] = e.extra.empty() ? nullptr : boost::json::value(e.extra);
boost::json::object req;
req["_route"] = "execute";
req["action"] = "insert";
req["table"] = "Logs";
req["values"] = vals;
client.call(boost::json::serialize(req), 1.0);
}
} catch (const std::exception& e) {
std::cerr << "[captcha2/log] batch drop: " << e.what() << std::endl;
}
}
std::string db_host_;
int db_port_;
std::atomic<int> level_threshold_{1}; // 日志级别阈值，默认 INFO(1)
std::unique_ptr<pmc::net::AsyncQueue<LogEntry>> queue_;
};
} // namespace captcha2
captcha2/src/main.cpp
+10 -27
View File
@@ -36,12 +36,12 @@
#include <unordered_map>
#include <vector>
#include "LogC.hpp"
#include <us/LogC.hpp>
namespace json = boost::json;
namespace po = boost::program_options;
using namespace captcha2;
using qing::LogC;
// ═══════════════════════════════════════════════════════════════════════════
// 1. 配置结构
@@ -70,13 +70,12 @@ struct CaptchaConfig {
int rate_limit_per_min = 600; // P1-1: 全局限流仅作最后防线，主要限流由 gateway 按 IP 负责
};
// 从 confs 拉取指定名称的配置文件全文，阻塞等待 confs 就绪
static std::string fetch_config_from_confs(const std::string& confs_host, int confs_port,
const std::string& config_name) {
// 从 confs 拉取 captcha.ini 全文，阻塞等待 confs 就绪
static std::string fetch_config_from_confs(const std::string& confs_host, int confs_port) {
pmc::net::RpcClient client(confs_host, confs_port, "captcha2",
/*timeout=*/5.0, /*retry_on_refused=*/true);
try {
auto data = client.call("get", {{"name", config_name}});
auto data = client.call("get", {{"name", "captcha.ini"}});
auto* content = data.as_object().if_contains("content");
if (!content || !content->is_string()) {
std::cerr << "[captcha2] confs 返回的 captcha.ini 无 content 字段" << std::endl;
@@ -526,15 +525,13 @@ static CaptchaConfig g_config;
static CaptchaStore g_store;
static RateLimiter g_rate_limiter;
static auto& g_log = LogC::instance();
static const auto _log_init = (LogC::instance().set_service_name("captcha2-log"), true);
int main(int argc, char* argv[]) {
try {
// ── 命令行参数 ──
std::string confs_host = "127.0.0.1";
int confs_port = 8010;
std::string confs_config_name = "captcha.ini"; // --config 覆盖
std::string instance = "captcha2"; // --instance 覆盖
std::string log_level = "info"; // --log-level 覆盖
{
const char* env;
if ((env = std::getenv("CONFS_HOST"))) confs_host = env;
@@ -544,13 +541,9 @@ int main(int argc, char* argv[]) {
po::options_description desc("captcha2 — pymsg 验证码服务");
desc.add_options()
("help,h", "显示帮助")
("addr,a", po::value<std::string>(), "监听地址（覆盖 confs，别名 --host）")
("host", po::value<std::string>(), "监听地址（覆盖 confs，别名 --addr/-a）")
("addr,a", po::value<std::string>(), "监听地址（覆盖 confs）")
("port,p", po::value<int>(), "监听端口（覆盖 confs）")
("workers,w", po::value<int>(), "工作线程数（覆盖 confs）")
("config", po::value<std::string>(), "confs 拉取的配置名（默认 captcha.ini）")
("log-level", po::value<std::string>(), "日志级别 debug|info|warn|error，默认 info")
("instance", po::value<std::string>(), "实例名（注册中心用，默认 captcha2）");
("workers,w", po::value<int>(), "工作线程数（覆盖 confs）");
po::variables_map vm;
po::store(po::parse_command_line(argc, argv, desc), vm);
po::notify(vm);
@@ -560,30 +553,20 @@ int main(int argc, char* argv[]) {
return 0;
}
// 预扫描 --config / --instance / --log-level / --host（confs 拉取前处理）
if (vm.count("config")) confs_config_name = vm["config"].as<std::string>();
if (vm.count("instance")) instance = vm["instance"].as<std::string>();
if (vm.count("log-level")) log_level = vm["log-level"].as<std::string>();
// ── 配置加载：confs → env → 命令行 ──
std::string ini_text = fetch_config_from_confs(confs_host, confs_port, confs_config_name);
std::string ini_text = fetch_config_from_confs(confs_host, confs_port);
merge_ini(g_config, ini_text);
apply_env_overrides(g_config);
if (vm.count("addr")) g_config.host = vm["addr"].as<std::string>();
if (vm.count("host")) g_config.host = vm["host"].as<std::string>();
if (vm.count("port")) g_config.port = vm["port"].as<int>();
if (vm.count("workers")) g_config.workers = vm["workers"].as<int>();
validate_config(g_config);
g_log.set_level(log_level);
std::cerr << "[captcha2] 配置来源: confs " << confs_config_name << " @ "
std::cerr << "[captcha2] 配置来源: confs captcha.ini @ "
<< confs_host << ":" << confs_port << std::endl;
std::cerr << "[captcha2] instance=" << instance
<< " log-level=" << log_level << std::endl;
std::cerr << "[captcha2] captcha_len=" << g_config.len
<< " ttl=" << g_config.ttl
<< " max_store=" << g_config.max_store
confs/log_client.py
+2 -158
View File
@@ -1,160 +1,4 @@
"""
统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
用法（与旧版完全兼容）：
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "WARNING", "登录失败次数超限",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import os as _os
import sys as _sys
import atexit
from datetime import datetime
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前系统时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now().astimezone().isoformat()
class LogClient:
"""异步批量日志推送客户端（API 与旧版完全兼容）。"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(),
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
except Exception as e:
print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
break
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
# ── 全局单例 ──────────────────────────────────────────────────────────
# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
# 此文件（confs/log_client.py）的 service_name 为 "confs"。
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
logs = LogClient(service_name="confs")
confs2/Makefile
+1 -1
View File
@@ -1,7 +1,7 @@
CXX := g++
CXXFLAGS := -std=c++20 -Wall -Wextra -O2
INCS := -I./inc
LIBS := -lus_msg -lboost_json -lboost_program_options -lpthread
LIBS := -lus_msg -lus_logc -lboost_json -lboost_program_options -lpthread
TARGET := confs2.bin
SRC := src/main.cpp
confs2/inc/LogC.hpp
-129
View File
@@ -1,129 +0,0 @@
/*
* LogC.hpp — confs2 异步日志客户端
*
* 基于 libmsg AsyncQueue，异步批量推送日志到 db:8005 的 Logs 表。
* - 内存队列 + 后台线程攒批（满 50 条或 2 秒 flush）
* - fail-safe：db 不可用时静默丢弃
* - API: info() / warning() / error()
*
* 改编自 captcha2/inc/LogC.hpp，仅改 namespace 和服务名。
*/
#pragma once
#include <us/Msg.hpp>
#include <us/AsyncQueue.hpp>
#include <boost/json.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <cstdlib>
#include <algorithm>
#include <cctype>
namespace confs2 {
struct LogEntry {
std::string ts;
std::string service;
std::string level;
std::string message;
std::string trace_id;
std::string extra;
};
class LogC {
public:
static LogC& instance() {
static LogC inst;
return inst;
}
void info(const std::string& svc, const std::string& msg) {
push(svc, "INFO", msg);
}
void warning(const std::string& svc, const std::string& msg) {
push(svc, "WARNING", msg);
}
void error(const std::string& svc, const std::string& msg) {
push(svc, "ERROR", msg);
}
private:
LogC() {
const char* host = std::getenv("DB_HOST");
const char* port = std::getenv("DB_PORT");
db_host_ = host ? host : "127.0.0.1";
db_port_ = port ? std::atoi(port) : 8005;
queue_ = std::make_unique<pmc::net::AsyncQueue<LogEntry>>(
[this](std::vector<LogEntry>&& batch) { send_batch(std::move(batch)); },
pmc::net::AsyncQueueConfig{.batch_size = 50, .flush_interval = 2.0, .max_queue_size = 10000},
[](const std::exception& e, std::vector<LogEntry>&&) {
std::cerr << "[confs2/log] send failed: " << e.what() << std::endl;
});
queue_->start();
}
~LogC() { if (queue_) queue_->shutdown(); }
// P2-3 修复：过滤控制字符，防止日志注入
static std::string sanitize_for_log(std::string s) {
s.erase(std::remove_if(s.begin(), s.end(), [](unsigned char c) {
return c < 0x20 && c != '\t';
}), s.end());
return s;
}
void push(const std::string& svc, const std::string& level, const std::string& msg) {
auto now = std::chrono::system_clock::now();
auto t = std::chrono::system_clock::to_time_t(now);
// 系统本地时间 + 时区偏移
std::tm* lt = std::localtime(&t);
char buf[40];
std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S", lt);
char tz_raw[8];
std::strftime(tz_raw, sizeof(tz_raw), "%z", lt);
std::string tz = tz_raw;
if (tz.length() >= 5)
tz = tz.substr(0, 3) + ":" + tz.substr(3);
LogEntry e;
e.ts = std::string(buf) + tz;
e.service = sanitize_for_log(svc);
e.level = level;
e.message = sanitize_for_log(msg);
queue_->push(std::move(e));
}
void send_batch(std::vector<LogEntry>&& batch) {
try {
pmc::net::MsgClient client(db_host_, db_port_, "confs-log");
for (auto& e : batch) {
boost::json::object vals;
vals["ts"] = e.ts;
vals["service"] = e.service;
vals["level"] = e.level;
vals["message"] = e.message;
vals["trace_id"] = e.trace_id.empty() ? nullptr : boost::json::value(e.trace_id);
vals["extra"] = e.extra.empty() ? nullptr : boost::json::value(e.extra);
boost::json::object req;
req["_route"] = "execute";
req["action"] = "insert";
req["table"] = "Logs";
req["values"] = vals;
client.call(boost::json::serialize(req), 1.0);
}
} catch (const std::exception& e) {
std::cerr << "[confs2/log] batch drop: " << e.what() << std::endl;
}
}
std::string db_host_;
int db_port_;
std::unique_ptr<pmc::net::AsyncQueue<LogEntry>> queue_;
};
} // namespace confs2
confs2/src/main.cpp
+3 -2
View File
@@ -31,13 +31,13 @@
#include <unordered_set>
#include <vector>
#include "LogC.hpp"
#include <us/LogC.hpp>
namespace json = boost::json;
namespace po = boost::program_options;
namespace fs = std::filesystem;
using confs2::LogC;
using qing::LogC;
// ═══════════════════════════════════════════════════════════════════════
// 1. 常量与异常
@@ -411,6 +411,7 @@ static void validate_config(const ConfsConfig& cfg) {
static ConfsConfig g_config;
static ConfigStore* g_store = nullptr;
static auto& g_log = LogC::instance();
static const auto _log_init = (LogC::instance().set_service_name("confs2-log"), true);
static const auto g_start_time = std::chrono::steady_clock::now();
static std::int64_t uptime_seconds() {
core/Makefile
+1 -1
View File
@@ -16,7 +16,7 @@ CPP:= g++
INCS:= -I include/
MY_LIBS := -L/usr/lib -lus_th -lus_http -lus_httpclient -lus_msg -lcurl
MY_LIBS := -L/usr/lib -lus_th -lus_http -lus_httpclient -lus_msg -lus_logc -lcurl
LIBS := -lpthread -lssl -lcrypto -lboost_program_options -lboost_json -lboost_thread -lboost_filesystem -lyaml-cpp
CPPFLAGS := --std=c++20 -O3 -pipe #-Wall -Werror
core/include/main/ConfsLogC.hpp
-132
View File
@@ -1,132 +0,0 @@
/*
* ConfsLogC.hpp — core 内嵌 confs 服务的异步日志客户端
*
* 基于 libmsg AsyncQueue，异步批量推送日志到 db:8005 的 Logs 表。
* - 内存队列 + 后台线程攒批（满 50 条或 2 秒 flush）
* - fail-safe：db 不可用时静默丢弃
* - API: info() / warning() / error()
*
* 改编自 confs2/inc/LogC.hpp，仅改 namespace（qing::confs），
* 服务名保持 "confs" 与 Python/standalone confs2 日志连续。
*/
#pragma once
#include <us/Msg.hpp>
#include <us/AsyncQueue.hpp>
#include <boost/json.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <cstdlib>
#include <algorithm>
#include <cctype>
namespace qing {
namespace confs {
struct ConfsLogEntry {
std::string ts;
std::string service;
std::string level;
std::string message;
std::string trace_id;
std::string extra;
};
class ConfsLogC {
public:
static ConfsLogC& instance() {
static ConfsLogC inst;
return inst;
}
void info(const std::string& svc, const std::string& msg) {
push(svc, "INFO", msg);
}
void warning(const std::string& svc, const std::string& msg) {
push(svc, "WARNING", msg);
}
void error(const std::string& svc, const std::string& msg) {
push(svc, "ERROR", msg);
}
private:
ConfsLogC() {
const char* host = std::getenv("DB_HOST");
const char* port = std::getenv("DB_PORT");
db_host_ = host ? host : "127.0.0.1";
db_port_ = port ? std::atoi(port) : 8005;
queue_ = std::make_unique<pmc::net::AsyncQueue<ConfsLogEntry>>(
[this](std::vector<ConfsLogEntry>&& batch) { send_batch(std::move(batch)); },
pmc::net::AsyncQueueConfig{.batch_size = 50, .flush_interval = 2.0, .max_queue_size = 10000},
[](const std::exception& e, std::vector<ConfsLogEntry>&&) {
std::cerr << "[core/confs-log] send failed: " << e.what() << std::endl;
});
queue_->start();
}
~ConfsLogC() { if (queue_) queue_->shutdown(); }
// P2-3 修复：过滤控制字符，防止日志注入
static std::string sanitize_for_log(std::string s) {
s.erase(std::remove_if(s.begin(), s.end(), [](unsigned char c) {
return c < 0x20 && c != '\t';
}), s.end());
return s;
}
void push(const std::string& svc, const std::string& level, const std::string& msg) {
auto now = std::chrono::system_clock::now();
auto t = std::chrono::system_clock::to_time_t(now);
// 系统本地时间 + 时区偏移
std::tm* lt = std::localtime(&t);
char buf[40];
std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S", lt);
char tz_raw[8];
std::strftime(tz_raw, sizeof(tz_raw), "%z", lt);
std::string tz = tz_raw;
if (tz.length() >= 5)
tz = tz.substr(0, 3) + ":" + tz.substr(3);
ConfsLogEntry e;
e.ts = std::string(buf) + tz;
e.service = sanitize_for_log(svc);
e.level = level;
e.message = sanitize_for_log(msg);
queue_->push(std::move(e));
}
void send_batch(std::vector<ConfsLogEntry>&& batch) {
try {
pmc::net::MsgClient client(db_host_, db_port_, "confs-log");
for (auto& e : batch) {
boost::json::object vals;
vals["ts"] = e.ts;
vals["service"] = e.service;
vals["level"] = e.level;
vals["message"] = e.message;
vals["trace_id"] = e.trace_id.empty() ? nullptr : boost::json::value(e.trace_id);
vals["extra"] = e.extra.empty() ? nullptr : boost::json::value(e.extra);
boost::json::object req;
req["_route"] = "execute";
req["action"] = "insert";
req["table"] = "Logs";
req["values"] = vals;
client.call(boost::json::serialize(req), 1.0);
}
} catch (const std::exception& e) {
std::cerr << "[core/confs-log] batch drop: " << e.what() << std::endl;
}
}
std::string db_host_;
int db_port_;
std::unique_ptr<pmc::net::AsyncQueue<ConfsLogEntry>> queue_;
};
} // namespace confs
} // namespace qing
core/src/main/ConfsTask.cpp
+9 -8
View File
@@ -31,12 +31,12 @@
#include <vector>
#include "main/ConfsTask.hh"
#include "main/ConfsLogC.hpp"
#include <us/LogC.hpp>
namespace json = boost::json;
namespace fs = std::filesystem;
using qing::confs::ConfsLogC;
using qing::LogC;
namespace qing {
@@ -386,6 +386,7 @@ ConfsTask::ConfsTask(std::string addr_, int port_, int workers_, std::string con
: addr(std::move(addr_)), port(port_), workers(workers_),
started_at(std::chrono::steady_clock::now())
{
LogC::instance().set_service_name("confs-log");
store = std::make_unique<ConfigStore>(config_dir);
this->th = std::make_unique<Thread>();
init_thread();
@@ -423,12 +424,12 @@ void ConfsTask::init_thread() {
this->server = std::make_unique<pmc::net::RpcServer> (addr, port,
pmc::net::MsgServerConfig{.workers = workers},
[](const std::string& route, const std::exception& e) {
ConfsLogC::instance().error("confs", "路由 " + route + " 处理异常: " + e.what());
LogC::instance().error("confs", "路由 " + route + " 处理异常: " + e.what());
});
register_routes(*this->server);
th.run();
ConfsLogC::instance().info("confs", "开始监听 tcp://" + addr + ":"
LogC::instance().info("confs", "开始监听 tcp://" + addr + ":"
+ std::to_string(port));
this->server->start(); /* 堵死线程，等待 stop() 解除 */
});
@@ -482,13 +483,13 @@ void ConfsTask::register_routes(pmc::net::RpcServer& rpc) {
auto val = extract_ini_key(entry.content, key);
if (!val)
throw KeyNotFound("键不存在: " + name + " -> " + key);
ConfsLogC::instance().info("confs", "分发配置键 " + name + " -> " + key);
LogC::instance().info("confs", "分发配置键 " + name + " -> " + key);
return json::object{{"name", name}, {"key", key}, {"value", *val}};
}
if (ext == ".json") {
auto val = extract_json_key(entry.content, key);
ConfsLogC::instance().info("confs", "分发配置键 " + name + " -> " + key);
LogC::instance().info("confs", "分发配置键 " + name + " -> " + key);
return json::object{{"name", name}, {"key", key}, {"value", val}};
}
@@ -519,7 +520,7 @@ void ConfsTask::register_routes(pmc::net::RpcServer& rpc) {
// 省略 name 或 null → 清空全部缓存
if (it == req.end() || it->value().is_null()) {
auto names = store->reload(nullptr);
ConfsLogC::instance().info("confs", "清空配置缓存（" + std::to_string(names.size()) + " 项）");
LogC::instance().info("confs", "清空配置缓存（" + std::to_string(names.size()) + " 项）");
json::array arr;
for (const auto& n : names) arr.push_back(json::value(n));
return json::object{{"reloaded", arr}};
@@ -534,7 +535,7 @@ void ConfsTask::register_routes(pmc::net::RpcServer& rpc) {
throw ConfigError("name 字段不能为空（要清空全部缓存请省略 name 字段）");
auto names = store->reload(&name);
ConfsLogC::instance().info("confs", "重载配置 " + names[0]);
LogC::instance().info("confs", "重载配置 " + names[0]);
json::array arr;
for (const auto& n : names) arr.push_back(json::value(n));
return json::object{{"reloaded", arr}};
dnfs/log_client.py
+2 -158
View File
@@ -1,160 +1,4 @@
"""
统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
用法（与旧版完全兼容）：
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "WARNING", "登录失败次数超限",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import os as _os
import sys as _sys
import atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
class LogClient:
"""异步批量日志推送客户端（API 与旧版完全兼容）。"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(),
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
except Exception as e:
print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
break
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
# ── 全局单例 ──────────────────────────────────────────────────────────
# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
# 此文件（confs/log_client.py）的 service_name 为 "confs"。
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
logs = LogClient(service_name="dnfs")
gateway/Makefile
+2 -2
View File
@@ -3,7 +3,7 @@ APP_NAME := gateway
INSTL_DIR := /usr/local/bin
CPP := g++
INCS := -I inc/ -I ../libmsg/inc
MY_LIBS := -L/usr/lib -L../libmsg/build -lus_http -lus_httpclient -lus_crypt -lus_msg
MY_LIBS := -L/usr/lib -L../libmsg/build -lus_http -lus_httpclient -lus_crypt -lus_msg -lus_logc
LIBS := -pthread -lcurl -lssl -lz -lcrypto -lboost_json -lboost_program_options -Wl,-rpath,$(shell cd ../libmsg/build && pwd)
CPPFLAGS := --std=c++20 -pipe -pthread # -Wall -Werror
BUILD_DIR = build
@@ -13,7 +13,7 @@ $(shell mkdir -p $(BUILD_DIR))
all: $(BUILD_DIR)/$(APP_NAME).out
$(BUILD_DIR)/$(APP_NAME).out: $(BUILD_DIR)/main.o
$(CPP) $^ $(LIBS) $(MY_LIBS) -o $@
$(BUILD_DIR)/main.o: src/main.cpp inc/LogC.hpp inc/Conf.hpp inc/ExpBase.hpp
$(BUILD_DIR)/main.o: src/main.cpp inc/Conf.hpp inc/ExpBase.hpp
$(CPP) $(CPPFLAGS) -c $< $(INCS) -o $@
clear:
@rm -rvf build/
gateway/inc/LogC.hpp
-235
View File
@@ -1,235 +0,0 @@
#pragma once
/** LogC — gateway 统一日志客户端（C++ 版，对应各 Python 服务的 log_client.py）
*
* 设计要点（与 Python 版对齐）：
* - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
* - 异步非阻塞：基于 pmc::net::AsyncQueue 内存队列 + 后台线程批量推送，不拖慢业务
* - fail-safe：db 服务不可用时静默丢弃（仅 stderr 提示），绝不影响业务
* - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
* - 背压保护：队列上限 10000 条，满时丢最旧日志，防止内存无限增长
*
* 用法：
* #include "LogC.hpp"
*
* qing::LogC::instance().info("gateway", "启动完成");
* qing::LogC::instance().error("gateway", "数据库连接失败",
* std::nullopt, R"({"errno":2002})");
*
* // 进程退出前：
* qing::LogC::instance().shutdown(); // 停止后台线程并 flush 剩余日志（幂等）
*
* // 设置日志级别（main 启动时调用，低于此级别的日志直接丢弃）
* qing::LogC::instance().set_level("warn"); // 只输出 warn/error
*/
#include <atomic>
#include <chrono>
#include <cstdio>
#include <ctime>
#include <mutex>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
#include <boost/json.hpp>
#include <us/AsyncQueue.hpp>
#include <us/Msg.hpp>
namespace qing {
namespace json = boost::json;
class LogC {
public:
struct Entry {
std::string service;
std::string level;
std::string message;
std::optional<std::string> trace_id;
std::optional<std::string> extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT）
};
static LogC& instance() {
static LogC inst;
return inst;
}
// ---- 公开接口 --------------------------------------------------------
// 日志级别 → 数值（大写兼容）
static const std::unordered_map<std::string, int>& level_order() {
static const std::unordered_map<std::string, int> k = {
{"debug", 0}, {"info", 1}, {"warn", 2}, {"warning", 2},
{"error", 3}, {"critical", 3},
{"DEBUG", 0}, {"INFO", 1}, {"WARN", 2}, {"WARNING", 2},
{"ERROR", 3}, {"CRITICAL", 3},
};
return k;
}
// 设置日志级别阈值（debug|info|warn|error，大小写不限）
// 低于此级别的日志在 push 时直接丢弃
void set_level(const std::string& level) {
auto it = level_order().find(level);
if (it != level_order().end()) {
_level_threshold.store(it->second, std::memory_order_relaxed);
}
}
// 推送一条日志（非阻塞，立即返回）
void push(const std::string& service, const std::string& level,
const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
// 日志级别过滤：低于阈值直接丢弃
auto it = level_order().find(level);
int lvl = (it != level_order().end()) ? it->second : 1;
if (lvl < _level_threshold.load(std::memory_order_relaxed)) return;
_queue.push(Entry{service, level, message,
std::move(trace_id), std::move(extra)});
}
void debug(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "DEBUG", message, std::move(trace_id), std::move(extra));
}
void info(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "INFO", message, std::move(trace_id), std::move(extra));
}
void warning(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "WARNING", message, std::move(trace_id), std::move(extra));
}
void error(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "ERROR", message, std::move(trace_id), std::move(extra));
}
void critical(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "CRITICAL", message, std::move(trace_id), std::move(extra));
}
// 手动 flush：通知后台线程立即发送当前积攒的批次，并等待其完成
void flush() {
_queue.flush();
}
// 停止后台线程并 flush 剩余日志（进程退出前调用，幂等）
void shutdown() {
_queue.shutdown();
}
// ---- 配置（构造后、首次推送前可改）------------------------------------
void set_endpoint(const std::string& addr, int port) {
std::lock_guard<std::mutex> lock(_endpoint_mtx);
_db_addr = addr;
_db_port = port;
}
private:
LogC()
: _queue(
[this](std::vector<Entry>&& batch) { this->_send_batch(std::move(batch)); },
{_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5},
[](const std::exception& e, std::vector<Entry>&&) {
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
})
{
_queue.start();
}
~LogC() { shutdown(); }
LogC(const LogC&) = delete;
LogC& operator=(const LogC&) = delete;
// 通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表
void _send_batch(const std::vector<Entry>& batch) {
std::string addr;
int port;
{
std::lock_guard<std::mutex> lock(_endpoint_mtx);
addr = _db_addr;
port = _db_port;
}
const std::string now = _utc_now_iso(); // 批次共享同一时间戳（对齐 Python 版）
try {
pmc::net::MsgClient client(addr, port, "gateway-log");
for (const auto& e : batch) {
json::object values;
values["ts"] = now;
values["service"] = e.service;
values["level"] = e.level;
values["message"] = e.message;
values["trace_id"] = e.trace_id ? json::value(*e.trace_id)
: json::value(nullptr);
values["extra"] = e.extra ? json::value(*e.extra)
: json::value(nullptr);
json::object payload;
payload["_route"] = "execute";
payload["action"] = "insert";
payload["table"] = "Logs";
payload["values"] = std::move(values);
const std::string resp = client.call(
json::serialize(payload), _TIMEOUT,
/*retry_on_refused=*/false); // 日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞重试
auto obj = json::parse(resp).as_object();
if (!obj.at("success").as_bool()) {
std::string err = "unknown";
if (auto* ev = obj.if_contains("error")) {
if (ev->is_string()) err = ev->as_string().c_str();
}
std::cerr << "[LogC] INSERT 失败: " << err << std::endl;
}
}
} catch (const std::exception& e) {
// fail-safe：db 不可用时静默丢弃，绝不影响业务
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
}
}
// UTC ISO8601 时间戳（对齐 Python datetime.now(timezone.utc).isoformat()）
// 形如 2026-07-29T08:30:00.123456+00:00，字典序即时间序，便于 db 按 ts 清理
static std::string _utc_now_iso() {
const auto now = std::chrono::system_clock::now();
const auto t = std::chrono::system_clock::to_time_t(now);
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
now.time_since_epoch()).count() % 1000000;
std::tm tm{};
::gmtime_r(&t, &tm);
char buf[40];
std::snprintf(buf, sizeof buf,
"%04d-%02d-%02dT%02d:%02d:%02d.%06d+00:00",
tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
tm.tm_hour, tm.tm_min, tm.tm_sec, static_cast<int>(us));
return buf;
}
// ---- 配置项 -----------------------------------------------------------
std::string _db_addr = "127.0.0.1";
int _db_port = 8005;
std::mutex _endpoint_mtx; // 保护 _db_addr/_db_port 读写（set_endpoint 与后台线程并发）
std::atomic<int> _level_threshold{1}; // 日志级别阈值，默认 INFO(1)
static constexpr size_t _BATCH_SIZE = 50; // 队列满 50 条时 flush
static constexpr double _FLUSH_INTERVAL = 2.0; // 超时 2 秒强制 flush
static constexpr size_t _MAX_QUEUE_SIZE = 10000; // 队列上限，满时丢最旧
static constexpr double _TIMEOUT = 5.0; // socket 超时（秒）
// ---- 核心队列 ---------------------------------------------------------
pmc::net::AsyncQueue<Entry> _queue;
};
} // namespace qing
gateway/src/main.cpp
+3 -1
View File
@@ -14,7 +14,8 @@
#include <boost/program_options.hpp>
#include "ExpBase.hpp"
#include "Conf.hpp"
#include "LogC.hpp"
#include <us/LogC.hpp>
#include <us/Msg.hpp>
using namespace pmc::net;
using namespace qing;
@@ -319,6 +320,7 @@ int main(int argc, char **argv) {
/* 初始化日志推送端点 */
logs().set_endpoint(_DB_HOST, _DB_PORT);
logs().set_service_name("gateway-log");
logs().set_level(_LOG_LEVEL);
std::cout << "[gateway] 监听: " << _ADDR << ":" << _PORT
liblogc/.gitignore
+1
View File
@@ -0,0 +1 @@
build/
liblogc/LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 建昌风物大顺脚本
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
liblogc/Makefile
+53
View File
@@ -0,0 +1,53 @@
# liblogc — 统一集中式日志客户端库
#
# 编译产物：libus_logc.so
# 依赖：libus_msg（libmsg）、boost_json
# 安装：头文件 → /usr/include/us/LogC.hpp
# 库文件 → /lib/libus_logc.so
CC := g++
STD := -std=c++20
VERSION := 0.0.1
FRAMEWORK_NAME := us
LIB_NAME := libus_logc
INSTL_TDIR := /lib
INCLU_TDIR := /usr/include
CFLAGS := $(STD) -O2 -fPIC -Wall
LDFLAGS := -shared -lus_msg -lboost_json
SRC_DIR := src
BUILD_DIR := build
INC_DIR := inc
SRC := $(wildcard $(SRC_DIR)/*.cpp)
OBJ := $(SRC:$(SRC_DIR)/%.cpp=$(BUILD_DIR)/%.o)
.PHONY: clear all install uninstall
all: $(BUILD_DIR) $(BUILD_DIR)/$(LIB_NAME).so
install: $(BUILD_DIR)/$(LIB_NAME).so
mkdir -p $(INCLU_TDIR)/$(FRAMEWORK_NAME)
cp -r $(INC_DIR)/$(FRAMEWORK_NAME)/* $(INCLU_TDIR)/$(FRAMEWORK_NAME)
cp $(BUILD_DIR)/$(LIB_NAME).so $(INSTL_TDIR)/$(LIB_NAME).so.$(VERSION)
ln -sf $(LIB_NAME).so.$(VERSION) $(INSTL_TDIR)/$(LIB_NAME).so
ldconfig
uninstall:
rm -f $(INCLU_TDIR)/$(FRAMEWORK_NAME)/LogC.hpp
rm -f $(INSTL_TDIR)/$(LIB_NAME).so
rm -f $(INSTL_TDIR)/$(LIB_NAME).so.$(VERSION)
clear:
rm -rf $(BUILD_DIR)
$(BUILD_DIR):
mkdir -p $(BUILD_DIR)
$(BUILD_DIR)/$(LIB_NAME).so: $(OBJ)
$(CC) $(OBJ) -o $@ $(LDFLAGS)
$(BUILD_DIR)/%.o: $(SRC_DIR)/%.cpp
$(CC) $(CFLAGS) -I$(INC_DIR) -c $< -o $@
liblogc/_set.sh
Executable
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
cd "$(dirname "$0")" || exit
make clear
make
make install
liblogc/inc/us/LogC.hpp
+144
View File
@@ -0,0 +1,144 @@
#pragma once
/** LogC — 统一集中式日志客户端（C++ 版）
*
* 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。
*
* 设计要点：
* - 异步非阻塞：基于 pmc::net::AsyncQueue 内存队列 + 后台线程批量推送
* - fail-safe：db 服务不可用时静默丢弃（仅 stderr 提示），绝不影响业务
* - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
* - 背压保护：队列上限 10000 条，满时丢最旧日志，防止内存无限增长
* - 级别过滤：set_level() 设定阈值，低于阈值的日志直接丢弃
* - 可配置 endpoint / service_name，各服务复用同一份库
*
* 用法：
* #include <us/LogC.hpp>
*
* // 启动时配置（默认 127.0.0.1:8005）
* qing::LogC::instance().set_endpoint("127.0.0.1", 8005);
* qing::LogC::instance().set_service_name("gateway-log");
* qing::LogC::instance().set_level("warn");
*
* qing::LogC::instance().info("gateway", "启动完成");
* qing::LogC::instance().error("gateway", "连接失败",
* std::nullopt, R"({"errno":111})");
*
* // 进程退出前调用（幂等）
* qing::LogC::instance().shutdown();
*/
#include <atomic>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
// forward-declare: 避免在头文件中拉入 AsyncQueue 模板
namespace pmc { namespace net {
template <typename T> class AsyncQueue;
}}
namespace qing {
class LogC {
public:
struct Entry {
std::string service;
std::string level;
std::string message;
std::optional<std::string> trace_id;
std::optional<std::string> extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT）
};
// 日志级别 → 数值（大写兼容）
static const std::unordered_map<std::string, int>& level_order() {
static const std::unordered_map<std::string, int> k = {
{"debug", 0}, {"info", 1}, {"warn", 2}, {"warning", 2},
{"error", 3}, {"critical", 3},
{"DEBUG", 0}, {"INFO", 1}, {"WARN", 2}, {"WARNING", 2},
{"ERROR", 3}, {"CRITICAL", 3},
};
return k;
}
static LogC& instance() {
static LogC inst;
return inst;
}
// ---- 配置（启动时调用，线程安全）-------------------------------------
void set_endpoint(const std::string& addr, int port);
void set_service_name(const std::string& name);
// 设置日志级别阈值（debug|info|warn|error，大小写不限）
void set_level(const std::string& level);
// ---- 推送接口（非阻塞，立即返回）-------------------------------------
void push(const std::string& service, const std::string& level,
const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt);
void debug(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "DEBUG", message, std::move(trace_id), std::move(extra));
}
void info(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "INFO", message, std::move(trace_id), std::move(extra));
}
void warning(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "WARNING", message, std::move(trace_id), std::move(extra));
}
void error(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "ERROR", message, std::move(trace_id), std::move(extra));
}
void critical(const std::string& service, const std::string& message,
std::optional<std::string> trace_id = std::nullopt,
std::optional<std::string> extra = std::nullopt) {
push(service, "CRITICAL", message, std::move(trace_id), std::move(extra));
}
// ---- 生命周期 -------------------------------------------------------
void flush(); // 手动 flush 当前积攒批次，等待完成
void shutdown(); // 停止后台线程并 flush 剩余日志（幂等）
private:
LogC();
~LogC();
LogC(const LogC&) = delete;
LogC& operator=(const LogC&) = delete;
void _send_batch(std::vector<Entry>&& batch);
static std::string _utc_now_iso();
// ---- 配置项 ---------------------------------------------------------
std::string _db_addr = "127.0.0.1";
int _db_port = 8005;
std::string _service_name = "log";
std::mutex _endpoint_mtx; // 保护 _db_addr/_db_port/_service_name
std::atomic<int> _level_threshold{1}; // 默认 INFO(1)
static constexpr size_t _BATCH_SIZE = 50;
static constexpr double _FLUSH_INTERVAL = 2.0;
static constexpr size_t _MAX_QUEUE_SIZE = 10000;
static constexpr double _TIMEOUT = 5.0;
// ---- 核心队列（unique_ptr 延迟到 .cpp 中实例化，保持头文件轻量）----
std::unique_ptr<pmc::net::AsyncQueue<Entry>> _queue;
};
} // namespace qing
liblogc/src/LogC.cpp
+141
View File
@@ -0,0 +1,141 @@
/** LogC.cpp — 统一集中式日志客户端实现
*
* 包含 _send_batch（pymsg → db execute INSERT）和 _utc_now_iso（ISO 8601）。
* 编译为 libus_logc.so，运行时依赖 libus_msg.so。
*/
#include <us/LogC.hpp>
#include <chrono>
#include <cstdio>
#include <ctime>
#include <iostream>
#include <memory>
#include <boost/json.hpp>
#include <us/AsyncQueue.hpp>
#include <us/Msg.hpp>
namespace json = boost::json;
namespace qing {
// ---- 生命周期 ---------------------------------------------------------------
LogC::LogC()
: _queue(std::make_unique<pmc::net::AsyncQueue<Entry>>(
[this](std::vector<Entry>&& batch) { this->_send_batch(std::move(batch)); },
pmc::net::AsyncQueueConfig{_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5},
[](const std::exception& e, std::vector<Entry>&&) {
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
}))
{
_queue->start();
}
LogC::~LogC() { shutdown(); }
// ---- 配置接口 ---------------------------------------------------------------
void LogC::set_endpoint(const std::string& addr, int port) {
std::lock_guard<std::mutex> lock(_endpoint_mtx);
_db_addr = addr;
_db_port = port;
}
void LogC::set_service_name(const std::string& name) {
std::lock_guard<std::mutex> lock(_endpoint_mtx);
_service_name = name;
}
void LogC::set_level(const std::string& level) {
auto it = level_order().find(level);
if (it != level_order().end()) {
_level_threshold.store(it->second, std::memory_order_relaxed);
}
}
// ---- 推送 -------------------------------------------------------------------
void LogC::push(const std::string& service, const std::string& level,
const std::string& message,
std::optional<std::string> trace_id,
std::optional<std::string> extra) {
auto it = level_order().find(level);
int lvl = (it != level_order().end()) ? it->second : 1;
if (lvl < _level_threshold.load(std::memory_order_relaxed)) return;
_queue->push(Entry{service, level, message,
std::move(trace_id), std::move(extra)});
}
void LogC::flush() { _queue->flush(); }
void LogC::shutdown() { _queue->shutdown(); }
// ---- 内部实现 ---------------------------------------------------------------
void LogC::_send_batch(std::vector<Entry>&& batch) {
std::string addr;
int port;
std::string client_name;
{
std::lock_guard<std::mutex> lock(_endpoint_mtx);
addr = _db_addr;
port = _db_port;
client_name = _service_name;
}
const std::string now = _utc_now_iso();
try {
pmc::net::MsgClient client(addr, port, client_name);
for (const auto& e : batch) {
json::object values;
values["ts"] = now;
values["service"] = e.service;
values["level"] = e.level;
values["message"] = e.message;
values["trace_id"] = e.trace_id ? json::value(*e.trace_id)
: json::value(nullptr);
values["extra"] = e.extra ? json::value(*e.extra)
: json::value(nullptr);
json::object payload;
payload["_route"] = "execute";
payload["action"] = "insert";
payload["table"] = "Logs";
payload["values"] = std::move(values);
const std::string resp = client.call(
json::serialize(payload), _TIMEOUT,
/*retry_on_refused=*/false);
auto obj = json::parse(resp).as_object();
if (!obj.at("success").as_bool()) {
std::string err = "unknown";
if (auto* ev = obj.if_contains("error")) {
if (ev->is_string()) err = ev->as_string().c_str();
}
std::cerr << "[LogC] INSERT 失败: " << err << std::endl;
}
}
} catch (const std::exception& e) {
// fail-safe：db 不可用时静默丢弃，绝不影响业务
std::cerr << "[LogC] 发送失败: " << e.what() << std::endl;
}
}
std::string LogC::_utc_now_iso() {
const auto now = std::chrono::system_clock::now();
const auto t = std::chrono::system_clock::to_time_t(now);
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
now.time_since_epoch()).count() % 1000000;
std::tm tm{};
::gmtime_r(&t, &tm);
char buf[40];
std::snprintf(buf, sizeof buf,
"%04d-%02d-%02dT%02d:%02d:%02d.%06d+00:00",
tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
tm.tm_hour, tm.tm_min, tm.tm_sec, static_cast<int>(us));
return buf;
}
} // namespace qing
pymsg/pymsg/__init__.py
+3 -1
View File
@@ -621,5 +621,7 @@ class MsgServer:
# ── JSON RPC 约定层（依赖上面的 MsgClient / MsgServer，故放在末尾导入）──
from .rpc import RpcClient, RpcError, RpcServer # noqa: E402
from .async_queue import AsyncRpcQueue # noqa: E402
from .logc import LogClient, set_log_level # noqa: E402
__all__ += ["RpcClient", "RpcServer", "RpcError", "AsyncRpcQueue"]
__all__ += ["RpcClient", "RpcServer", "RpcError", "AsyncRpcQueue",
"LogClient", "set_log_level"]
pymsg/pymsg/logc.py
+189
View File
@@ -0,0 +1,189 @@
"""
pymsg.logc — 统一集中式日志客户端（Python 版）
通过 pymsg RPC 调用 db execute 路由，逐条 INSERT 到 Logs 表。
设计要点：
- 异步非阻塞：基于 pymsg.AsyncRpcQueue 内存队列 + 后台线程攒批推送
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 背压保护：队列上限 10000 条，满时丢最旧日志，防止 OOM
- 日志级别过滤：set_level() 设定阈值，低于阈值的日志直接丢弃
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
用法：
from pymsg.logc import LogClient, set_log_level
logs = LogClient(service_name="redirect80")
logs.info("redirect80", "服务已启动")
logs.error("redirect80", "端口绑定失败", extra={"errno": 98})
# 设置日志级别（main 启动时调用）
set_log_level("warn") # 只输出 warn/error/critical
"""
import atexit
import json
import os
import sys
from datetime import datetime, timezone
from pymsg import AsyncRpcQueue, RpcClient, RpcError
# ── 配置 ──────────────────────────────────────────────────────────────────────
_DBV2_ADDR = os.environ.get("DB_HOST", "127.0.0.1")
_DBV2_PORT = int(os.environ.get("DB_PORT", "8005"))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
# ── 日志级别过滤 ──────────────────────────────────────────────────────────────
_LEVEL_ORDER = {
"DEBUG": 0, "INFO": 1,
"WARNING": 2, "WARN": 2,
"ERROR": 3, "CRITICAL": 3,
}
_level_threshold = 1 # 默认 INFO
def set_log_level(level: str):
"""设置全局日志级别阈值。低于此级别的日志直接丢弃。"""
global _level_threshold
lv = level.upper()
if lv == "WARNING":
lv = "WARN"
_level_threshold = _LEVEL_ORDER.get(lv, 1)
# ── 工具函数 ──────────────────────────────────────────────────────────────────
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
# ── LogClient 类 ──────────────────────────────────────────────────────────────
class LogClient:
"""异步批量日志推送客户端。
API 与 redirect80/log_client.py、confs/log_client.py 等保持一致，
各服务可直接替换为 ``from pymsg.logc import LogClient``。
"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ──────────────────────────────────────────────────────────
def set_log_level(self, level: str):
"""设置日志过滤阈值（委托给模块级 set_log_level）。"""
set_log_level(level)
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
低于设定级别的日志直接丢弃。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程。
"""
if _LEVEL_ORDER.get(level.upper(), 1) < _level_threshold:
return
entry = {
"ts": _now_iso(),
"service": service,
"level": level,
"message": message,
"trace_id": trace_id,
"extra": extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, "DEBUG", message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, "INFO", message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, "WARNING", message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, "WARNING", message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, "ERROR", message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, "CRITICAL", message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ──────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db（网络错误放弃本批剩余）。"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
print(f"[logc] INSERT 失败: {e}", file=sys.stderr)
except Exception as e:
print(f"[logc] 发送失败（放弃本批剩余）: {e}", file=sys.stderr)
break
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[logc] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=sys.stderr)
redirect80/log_client.py
+2 -177
View File
@@ -1,179 +1,4 @@
"""
统一日志客户端 — redirect80 服务副本（基于 pymsg.AsyncRpcQueue）。
设计要点：
- 通过 pymsg RPC 调用 db execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
- 日志级别过滤：set_level() 设定阈值，低于阈值的日志直接丢弃
用法（与旧版完全兼容）：
from log_client import logs
logs.info("redirect80", "服务已启动")
logs.error("redirect80", "端口绑定失败", extra={"errno": 98})
logs.push("redirect80", "WARNING", "confs 不可用，使用本地配置",
trace_id="startup", extra={"fallback": "redirect80.ini"})
# 设置日志级别（main 启动时调用）
from log_client import set_log_level
set_log_level("warn") # 只输出 warn/error
"""
import json
import os as _os
import sys as _sys
import atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
# ── 日志级别过滤 ────────────────────────────────────────────────────
_LEVEL_ORDER = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "WARN": 2, "ERROR": 3, "CRITICAL": 3}
_level_threshold = 1 # 默认 INFO
def set_log_level(level: str):
"""设置全局日志级别阈值。低于此级别的日志直接丢弃。"""
global _level_threshold
lv = level.upper()
if lv == "WARNING":
lv = "WARN"
_level_threshold = _LEVEL_ORDER.get(lv, 1)
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
class LogClient:
"""异步批量日志推送客户端（API 与旧版完全兼容）。"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
低于设定级别的日志直接丢弃。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
if _LEVEL_ORDER.get(level.upper(), 1) < _level_threshold:
return
entry = {
'ts': _now_iso(),
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db（网络错误放弃本批剩余）。"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
except Exception as e:
print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
break
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
# ── 全局单例 ──────────────────────────────────────────────────────────
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
logs = LogClient(service_name="redirect80")
smtp/log_client.py
+3 -276
View File
@@ -1,277 +1,4 @@
"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts，反映事件真实发生时间
用法：
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "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 atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50 # 队列满 50 条时 flush
_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush
_TIMEOUT = 5 # socket 超时（秒）
_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
# ── 日志级别过滤 ──
_LOG_LEVELS = {
"debug": 0, "info": 1, "warn": 2, "warning": 2,
"error": 3, "critical": 4,
}
_LOG_LEVEL_THRESHOLD = _LOG_LEVELS["info"] # 默认 info
def _normalize_level(level):
"""归一化级别名：大写转小写，warning→warn。"""
lv = (level or "info").lower()
if lv == "warning":
lv = "warn"
return lv
def set_log_level(level):
"""设置日志过滤阈值（低于该级别的日志将被丢弃）。"""
global _LOG_LEVEL_THRESHOLD
lv = _normalize_level(level)
if lv in _LOG_LEVELS:
_LOG_LEVEL_THRESHOLD = _LOG_LEVELS[lv]
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
class LogClient:
"""异步批量日志推送客户端。"""
def __init__(self, service_name="log",
batch_size=_BATCH_SIZE,
flush_interval=_FLUSH_INTERVAL):
self._service_name = service_name
self._batch_size = batch_size
self._flush_interval = flush_interval
self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
self._stop = threading.Event()
self._flush_signal = threading.Event()
self._flush_done = threading.Event()
self._thread = threading.Thread(target=self._flush_loop,
daemon=True, name='log-client')
self._thread.start()
atexit.register(self._shutdown)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service, level, message, trace_id=None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
# 级别过滤：低于当前阈值的日志直接丢弃（不进队列）
if _LOG_LEVELS.get(_normalize_level(level), 0) < _LOG_LEVEL_THRESHOLD:
return
entry = {
'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
try:
self._queue.put_nowait(entry)
except queue.Full:
# 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
try:
self._queue.get_nowait()
except queue.Empty:
pass
try:
self._queue.put_nowait(entry)
except queue.Full:
pass # 极端竞争下仍失败则静默丢弃
def debug(self, service, message, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service, message, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service, message, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service, message, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service, message, **kw):
self.push(service, 'CRITICAL', message, **kw)
def set_log_level(self, level):
"""设置日志过滤阈值（低于该级别的日志将被丢弃）。
委托给模块级 set_log_level，统一通过 logs 单例对外暴露接口。
"""
set_log_level(level)
def flush(self):
"""手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
self._flush_done.clear()
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
# 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
self._flush_done.wait(timeout=self._flush_interval + 2.0)
# 主线程排空队列中剩余条目
self._drain_queue_to_db()
def _drain_queue_to_db(self):
"""排空主线程可见的队列中剩余条目，直接发送到 db。"""
batch = []
while True:
try:
batch.append(self._queue.get_nowait())
if len(batch) >= self._batch_size:
self._send(batch)
batch = []
except queue.Empty:
break
if batch:
self._send(batch)
# ── 内部实现 ────────────────────────────────────────────────────────
def _flush_loop(self):
"""后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
都不会死亡，避免日志队列无限增长导致 OOM。
"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
try:
timeout = self._flush_interval - (time.monotonic() - last_flush)
chk_timeout = max(0.1, min(timeout, 0.5))
try:
item = self._queue.get(timeout=chk_timeout)
batch.append(item)
while len(batch) < self._batch_size:
try:
batch.append(self._queue.get_nowait())
except queue.Empty:
break
except queue.Empty:
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 and batch:
try:
self._send(batch)
except Exception as e:
# _send 内部已有细粒度异常处理，此处是最终安全网
print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
batch = []
last_flush = time.monotonic()
if self._flush_signal.is_set():
self._flush_signal.clear()
self._flush_done.set()
except Exception as e:
# 最外层安全网：任何意外异常都不应杀死日志线程
print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
time.sleep(0.5) # 防止异常风暴导致 CPU 满载
# 退出前 flush 剩余
if batch:
try:
self._send(batch)
except Exception as e:
print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
self._flush_done.set()
def _send(self, batch):
"""通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
不抛异常给调用方。
"""
try:
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
except Exception as e:
print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
return
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
client.call("execute", payload, timeout=_TIMEOUT)
except RpcError as e:
# 业务错误（如表结构不符）：打印后继续下一条
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
except Exception as e:
# 网络错误：db 不可用，放弃本批剩余（fail-safe）
print(f"[log_client] 发送失败: {e}", file=_sys.stderr)
break
def _shutdown(self):
"""进程退出时 flush 剩余日志。"""
self._stop.set()
self.flush()
# ── 全局单例 ──────────────────────────────────────────────────────────
logs = LogClient()
logs = LogClient(service_name="smtp")
smtp3/Makefile
+1 -1
View File
@@ -1,7 +1,7 @@
CXX := g++
CXXFLAGS := -std=c++20 -Wall -Wextra -O2
INCS := -I./inc
LIBS := -lus_msg -lboost_json -lboost_program_options -lssl -lcrypto -lpthread
LIBS := -lus_msg -lus_logc -lboost_json -lboost_program_options -lssl -lcrypto -lpthread
TARGET := smtp3.bin
SRC := src/main.cpp
smtp3/inc/LogC.hpp
-124
View File
@@ -1,124 +0,0 @@
/*
* LogC.hpp — smtp3 异步日志客户端（简化版）
*
* 基于 libmsg AsyncQueue，异步批量推送日志到 db:8005 的 Logs 表。
* - 内存队列 + 后台线程攒批（满 50 条或 2 秒 flush）
* - fail-safe：db 不可用时静默丢弃
* - P3-5: 长连接复用（底层 openConnection + pmsg_send/recv），
* 不再每批新建 TCP 连接；连接故障时关闭重建，本批丢弃
* - API: info() / warning() / error()
*/
#pragma once
#include <us/Msg.hpp>
#include <us/AsyncQueue.hpp>
#include <boost/json.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <cstdlib>
#include <unistd.h>
namespace smtp3 {
struct LogEntry {
std::string ts;
std::string service;
std::string level;
std::string message;
std::string trace_id;
std::string extra;
};
class LogC {
public:
static LogC& instance() {
static LogC inst;
return inst;
}
void info(const std::string& svc, const std::string& msg) {
push(svc, "INFO", msg);
}
void warning(const std::string& svc, const std::string& msg) {
push(svc, "WARNING", msg);
}
void error(const std::string& svc, const std::string& msg) {
push(svc, "ERROR", msg);
}
private:
LogC() {
const char* host = std::getenv("DB_HOST");
const char* port = std::getenv("DB_PORT");
db_host_ = host ? host : "127.0.0.1";
db_port_ = port ? std::atoi(port) : 8005;
queue_ = std::make_unique<pmc::net::AsyncQueue<LogEntry>>(
[this](std::vector<LogEntry>&& batch) { send_batch(std::move(batch)); },
pmc::net::AsyncQueueConfig{.batch_size = 50, .flush_interval = 2.0, .max_queue_size = 10000},
[](const std::exception& e, std::vector<LogEntry>&&) {
std::cerr << "[smtp3/log] send failed: " << e.what() << std::endl;
});
queue_->start();
}
~LogC() {
if (queue_) queue_->shutdown();
if (fd_ >= 0) ::close(fd_);
}
void push(const std::string& svc, const std::string& level, const std::string& msg) {
auto now = std::chrono::system_clock::now();
auto t = std::chrono::system_clock::to_time_t(now);
char buf[32];
std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", std::gmtime(&t));
LogEntry e;
e.ts = buf;
e.service = svc;
e.level = level;
e.message = msg;
queue_->push(std::move(e));
}
// P3-5: 长连接复用——同一 fd 上逐条 pmsg_send + pmsg_recv（db 对每条
// execute 都回复，须消费以保持帧同步）；任一异常即关闭连接、丢弃本批
void send_batch(std::vector<LogEntry>&& batch) {
try {
if (fd_ < 0)
fd_ = pmc::net::detail::openConnection(db_host_, db_port_, 1.0);
for (auto& e : batch) {
boost::json::object vals;
vals["ts"] = e.ts;
vals["service"] = e.service;
vals["level"] = e.level;
vals["message"] = e.message;
vals["trace_id"] = e.trace_id.empty() ? nullptr : boost::json::value(e.trace_id);
vals["extra"] = e.extra.empty() ? nullptr : boost::json::value(e.extra);
boost::json::object req;
req["_route"] = "execute";
req["action"] = "insert";
req["table"] = "Logs";
req["values"] = vals;
// 与 MsgClient::call 线路格式一致：无前缀，逐条请求-应答
pmc::net::pmsg_send(fd_, boost::json::serialize(req));
(void)pmc::net::pmsg_recv(fd_, 1.0);
}
} catch (const std::exception& e) {
if (fd_ >= 0) { ::close(fd_); fd_ = -1; }
std::cerr << "[smtp3/log] batch drop: " << e.what() << std::endl;
}
}
std::string db_host_;
int db_port_;
std::unique_ptr<pmc::net::AsyncQueue<LogEntry>> queue_;
int fd_ = -1; // 长连接 fd（仅后台 flush 线程访问，无并发）
};
} // namespace smtp3
smtp3/src/main.cpp
+3 -2
View File
@@ -59,12 +59,12 @@
#include <sys/socket.h>
#include <unistd.h>
#include "LogC.hpp"
#include <us/LogC.hpp>
namespace json = boost::json;
namespace po = boost::program_options;
using namespace smtp3;
using qing::LogC;
// ═══════════════════════════════════════════════════════════════════════════
// 辅助: 安全的 stoi
@@ -716,6 +716,7 @@ static SendLimiter g_limiter;
static std::unordered_map<std::string, double> g_cooldown;
static std::mutex g_cooldown_mtx;
static auto& g_log = LogC::instance();
static const auto _log_init = (LogC::instance().set_service_name("smtp3-log"), true);
// ═══════════════════════════════════════════════════════════════════════════
// 8. main() — RpcServer 启动
static/log_client.py
+2 -160
View File
@@ -1,162 +1,4 @@
"""
统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
- 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施
用法（与旧版完全兼容）：
from log_client import logs
logs.info("dnfs", "节点新增成功 id=42")
logs.error("dnfs", "数据库连接失败", extra={"errno": 2002})
logs.push("gateway", "WARNING", "登录失败次数超限",
trace_id="req-abc", extra={"ip": "1.2.3.4"})
"""
import json
import os as _os
import sys as _sys
import atexit
from datetime import datetime, timezone
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402
# ── 配置 ──────────────────────────────────────────────────────────────
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
_BATCH_SIZE = 50
_FLUSH_INTERVAL = 2.0
_TIMEOUT = 5
_QUEUE_MAXSIZE = 10000
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
class LogClient:
"""异步批量日志推送客户端（API 与旧版完全兼容）。"""
def __init__(self, service_name: str = "log",
batch_size: int = _BATCH_SIZE,
flush_interval: float = _FLUSH_INTERVAL):
self._service_name = service_name
self._client = RpcClient(
_DBV2_ADDR, _DBV2_PORT,
service_name + "-log",
timeout=_TIMEOUT,
)
self._queue = AsyncRpcQueue(
handler=self._send_batch,
batch_size=batch_size,
flush_interval=flush_interval,
max_queue_size=_QUEUE_MAXSIZE,
on_error=self._on_error,
)
self._queue.start()
atexit.register(self.close)
# ── 公开接口 ────────────────────────────────────────────────────────
def push(self, service: str, level: str, message: str,
trace_id: str = None, extra=None):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(),
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
self._queue.put(entry)
def debug(self, service: str, message: str, **kw):
self.push(service, 'DEBUG', message, **kw)
def info(self, service: str, message: str, **kw):
self.push(service, 'INFO', message, **kw)
def warning(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def warn(self, service: str, message: str, **kw):
self.push(service, 'WARNING', message, **kw)
def error(self, service: str, message: str, **kw):
self.push(service, 'ERROR', message, **kw)
def critical(self, service: str, message: str, **kw):
self.push(service, 'CRITICAL', message, **kw)
def flush(self):
"""手动 flush 队列中剩余的日志。"""
self._queue.flush()
def close(self):
"""关闭客户端，flush 剩余日志。"""
self._queue.close()
# ── 内部实现 ────────────────────────────────────────────────────────
def _send_batch(self, batch: list):
"""后台线程回调：逐条 INSERT 到 db。
NEW-13：单条失败仅跳过该条，不再因网络异常放弃整批剩余日志。
RpcError 视为可重试的业务错误（如 db 临时拒绝），继续处理下一条。
"""
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
try:
self._client.call("execute", payload, timeout=_TIMEOUT)
except Exception as e:
print(f"[log_client] INSERT 失败（跳过本条）: {e}", file=_sys.stderr)
continue
def _on_error(self, e: Exception, batch: list):
"""异常安全网：handler 外的异常也不会崩溃线程。"""
print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
# ── 全局单例 ──────────────────────────────────────────────────────────
# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
# 此文件（confs/log_client.py）的 service_name 为 "confs"。
"""日志客户端 — 委托到 pymsg.logc 统一库。"""
from pymsg.logc import LogClient, set_log_level # noqa: F401
logs = LogClient(service_name="static")