From aa3741b8619b7293a1a956f6dd75f04339ad0931 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Mon, 10 Aug 2026 04:51:43 +0000 Subject: [PATCH] =?UTF-8?q?feat(liblogc):=20=E7=BB=9F=E4=B8=80=E9=9B=86?= =?UTF-8?q?=E4=B8=AD=E5=BC=8F=E6=97=A5=E5=BF=97=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E5=BA=93=20+=20=E5=85=A8=E9=87=8F=E8=BF=81=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增: - liblogc/: C++ 统一日志库 (libus_logc.so)，支持 set_endpoint/set_service_name/set_level - pymsg/pymsg/logc.py: Python 统一日志模块，基于 AsyncRpcQueue C++ 迁移 (删除 6 份本地 LogC.hpp 副本): - ca, gateway: + 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): 保持现状，自成体系 --- ca/Makefile | 4 +- ca/inc/LogC.hpp | 235 --------------------------- ca/src/main.cpp | 3 +- captcha/log_client.py | 158 +----------------- captcha2/Makefile | 2 +- captcha2/inc/LogC.hpp | 139 ---------------- captcha2/src/main.cpp | 37 ++--- confs/log_client.py | 160 +----------------- confs2/Makefile | 2 +- confs2/inc/LogC.hpp | 129 --------------- confs2/src/main.cpp | 5 +- core/Makefile | 2 +- core/include/main/ConfsLogC.hpp | 132 --------------- core/src/main/ConfsTask.cpp | 17 +- dnfs/log_client.py | 160 +----------------- gateway/Makefile | 4 +- gateway/inc/LogC.hpp | 235 --------------------------- gateway/src/main.cpp | 4 +- liblogc/.gitignore | 1 + liblogc/LICENSE | 21 +++ liblogc/Makefile | 53 ++++++ liblogc/_set.sh | 5 + liblogc/inc/us/LogC.hpp | 144 +++++++++++++++++ liblogc/src/LogC.cpp | 141 ++++++++++++++++ pymsg/pymsg/__init__.py | 4 +- pymsg/pymsg/logc.py | 189 ++++++++++++++++++++++ redirect80/log_client.py | 179 +------------------- smtp/log_client.py | 279 +------------------------------- smtp3/Makefile | 2 +- smtp3/inc/LogC.hpp | 124 -------------- smtp3/src/main.cpp | 5 +- static/log_client.py | 162 +------------------ 32 files changed, 608 insertions(+), 2129 deletions(-) delete mode 100644 ca/inc/LogC.hpp delete mode 100644 captcha2/inc/LogC.hpp delete mode 100644 confs2/inc/LogC.hpp delete mode 100644 core/include/main/ConfsLogC.hpp delete mode 100644 gateway/inc/LogC.hpp create mode 100644 liblogc/.gitignore create mode 100644 liblogc/LICENSE create mode 100644 liblogc/Makefile create mode 100755 liblogc/_set.sh create mode 100644 liblogc/inc/us/LogC.hpp create mode 100644 liblogc/src/LogC.cpp create mode 100644 pymsg/pymsg/logc.py delete mode 100644 smtp3/inc/LogC.hpp diff --git a/ca/Makefile b/ca/Makefile index 33aa32b..fc0bbe4 100644 --- a/ca/Makefile +++ b/ca/Makefile @@ -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/ diff --git a/ca/inc/LogC.hpp b/ca/inc/LogC.hpp deleted file mode 100644 index 021411f..0000000 --- a/ca/inc/LogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include -#include - -#include - -#include -#include - -namespace qing { - -namespace json = boost::json; - -class LogC { -public: - struct Entry { - std::string service; - std::string level; - std::string message; - std::optional trace_id; - std::optional extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT） - }; - - // 日志级别 → 数值（大写兼容） - static const std::unordered_map& level_order() { - static const std::unordered_map 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 lock(_endpoint_mtx); - _db_addr = addr; - _db_port = port; - } - -private: - LogC() - : _queue( - [this](std::vector&& batch) { this->_send_batch(std::move(batch)); }, - {_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5}, - [](const std::exception& e, std::vector&&) { - 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& batch) { - std::string addr; - int port; - { - std::lock_guard 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( - 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(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 _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 _queue; -}; - -} // namespace qing diff --git a/ca/src/main.cpp b/ca/src/main.cpp index d6e193c..aa163c6 100644 --- a/ca/src/main.cpp +++ b/ca/src/main.cpp @@ -25,7 +25,7 @@ #include "LocalTime.hpp" #include "Conf.hpp" #include "DbC.hpp" -#include "LogC.hpp" +#include 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); diff --git a/captcha/log_client.py b/captcha/log_client.py index 218ba3d..f192d69 100644 --- a/captcha/log_client.py +++ b/captcha/log_client.py @@ -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") diff --git a/captcha2/Makefile b/captcha2/Makefile index a1c0645..9e5ec14 100644 --- a/captcha2/Makefile +++ b/captcha2/Makefile @@ -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 diff --git a/captcha2/inc/LogC.hpp b/captcha2/inc/LogC.hpp deleted file mode 100644 index 67e201f..0000000 --- a/captcha2/inc/LogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include -#include -#include - -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& level_order() { - static const std::unordered_map 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>( - [this](std::vector&& 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&&) { - 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&& 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 level_threshold_{1}; // 日志级别阈值，默认 INFO(1) - std::unique_ptr> queue_; -}; - -} // namespace captcha2 diff --git a/captcha2/src/main.cpp b/captcha2/src/main.cpp index b54706e..fcb808d 100644 --- a/captcha2/src/main.cpp +++ b/captcha2/src/main.cpp @@ -36,12 +36,12 @@ #include #include -#include "LogC.hpp" +#include 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(), "监听地址（覆盖 confs，别名 --host）") - ("host", po::value(), "监听地址（覆盖 confs，别名 --addr/-a）") + ("addr,a", po::value(), "监听地址（覆盖 confs）") ("port,p", po::value(), "监听端口（覆盖 confs）") - ("workers,w", po::value(), "工作线程数（覆盖 confs）") - ("config", po::value(), "confs 拉取的配置名（默认 captcha.ini）") - ("log-level", po::value(), "日志级别 debug|info|warn|error，默认 info") - ("instance", po::value(), "实例名（注册中心用，默认 captcha2）"); + ("workers,w", po::value(), "工作线程数（覆盖 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(); - if (vm.count("instance")) instance = vm["instance"].as(); - if (vm.count("log-level")) log_level = vm["log-level"].as(); - // ── 配置加载：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(); - if (vm.count("host")) g_config.host = vm["host"].as(); if (vm.count("port")) g_config.port = vm["port"].as(); if (vm.count("workers")) g_config.workers = vm["workers"].as(); 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 diff --git a/confs/log_client.py b/confs/log_client.py index edc0b6e..a18cc22 100644 --- a/confs/log_client.py +++ b/confs/log_client.py @@ -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") diff --git a/confs2/Makefile b/confs2/Makefile index 996b3a7..fdc9683 100644 --- a/confs2/Makefile +++ b/confs2/Makefile @@ -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 diff --git a/confs2/inc/LogC.hpp b/confs2/inc/LogC.hpp deleted file mode 100644 index 40ab86b..0000000 --- a/confs2/inc/LogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include -#include -#include - -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>( - [this](std::vector&& 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&&) { - 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&& 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> queue_; -}; - -} // namespace confs2 diff --git a/confs2/src/main.cpp b/confs2/src/main.cpp index 92f89b4..8889a39 100644 --- a/confs2/src/main.cpp +++ b/confs2/src/main.cpp @@ -31,13 +31,13 @@ #include #include -#include "LogC.hpp" +#include 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() { diff --git a/core/Makefile b/core/Makefile index fd88b6f..7c1722e 100644 --- a/core/Makefile +++ b/core/Makefile @@ -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 diff --git a/core/include/main/ConfsLogC.hpp b/core/include/main/ConfsLogC.hpp deleted file mode 100644 index cd3d8d5..0000000 --- a/core/include/main/ConfsLogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include -#include -#include - -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>( - [this](std::vector&& 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&&) { - 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&& 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> queue_; -}; - -} // namespace confs -} // namespace qing diff --git a/core/src/main/ConfsTask.cpp b/core/src/main/ConfsTask.cpp index ce8f514..c54b858 100644 --- a/core/src/main/ConfsTask.cpp +++ b/core/src/main/ConfsTask.cpp @@ -31,12 +31,12 @@ #include #include "main/ConfsTask.hh" -#include "main/ConfsLogC.hpp" +#include 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(config_dir); this->th = std::make_unique(); init_thread(); @@ -423,12 +424,12 @@ void ConfsTask::init_thread() { this->server = std::make_unique (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}}; diff --git a/dnfs/log_client.py b/dnfs/log_client.py index e61f68a..6a2051e 100644 --- a/dnfs/log_client.py +++ b/dnfs/log_client.py @@ -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") diff --git a/gateway/Makefile b/gateway/Makefile index 2bad387..08f0b81 100644 --- a/gateway/Makefile +++ b/gateway/Makefile @@ -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/ diff --git a/gateway/inc/LogC.hpp b/gateway/inc/LogC.hpp deleted file mode 100644 index a77d23d..0000000 --- a/gateway/inc/LogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include -#include - -#include - -#include -#include - -namespace qing { - -namespace json = boost::json; - -class LogC { -public: - struct Entry { - std::string service; - std::string level; - std::string message; - std::optional trace_id; - std::optional extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT） - }; - - static LogC& instance() { - static LogC inst; - return inst; - } - - // ---- 公开接口 -------------------------------------------------------- - - // 日志级别 → 数值（大写兼容） - static const std::unordered_map& level_order() { - static const std::unordered_map 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 trace_id = std::nullopt, - std::optional 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 lock(_endpoint_mtx); - _db_addr = addr; - _db_port = port; - } - -private: - LogC() - : _queue( - [this](std::vector&& batch) { this->_send_batch(std::move(batch)); }, - {_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5}, - [](const std::exception& e, std::vector&&) { - 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& batch) { - std::string addr; - int port; - { - std::lock_guard 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( - 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(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 _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 _queue; -}; - -} // namespace qing diff --git a/gateway/src/main.cpp b/gateway/src/main.cpp index 3fd2bc9..685042e 100644 --- a/gateway/src/main.cpp +++ b/gateway/src/main.cpp @@ -14,7 +14,8 @@ #include #include "ExpBase.hpp" #include "Conf.hpp" -#include "LogC.hpp" +#include +#include 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 diff --git a/liblogc/.gitignore b/liblogc/.gitignore new file mode 100644 index 0000000..567609b --- /dev/null +++ b/liblogc/.gitignore @@ -0,0 +1 @@ +build/ diff --git a/liblogc/LICENSE b/liblogc/LICENSE new file mode 100644 index 0000000..d63a519 --- /dev/null +++ b/liblogc/LICENSE @@ -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. diff --git a/liblogc/Makefile b/liblogc/Makefile new file mode 100644 index 0000000..f18e91b --- /dev/null +++ b/liblogc/Makefile @@ -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 $@ diff --git a/liblogc/_set.sh b/liblogc/_set.sh new file mode 100755 index 0000000..312f67c --- /dev/null +++ b/liblogc/_set.sh @@ -0,0 +1,5 @@ +#!/bin/bash +cd "$(dirname "$0")" || exit +make clear +make +make install diff --git a/liblogc/inc/us/LogC.hpp b/liblogc/inc/us/LogC.hpp new file mode 100644 index 0000000..269f6a6 --- /dev/null +++ b/liblogc/inc/us/LogC.hpp @@ -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 + * + * // 启动时配置（默认 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 +#include +#include +#include +#include +#include +#include + +// forward-declare: 避免在头文件中拉入 AsyncQueue 模板 +namespace pmc { namespace net { +template class AsyncQueue; +}} + +namespace qing { + +class LogC { +public: + struct Entry { + std::string service; + std::string level; + std::string message; + std::optional trace_id; + std::optional extra; // 已序列化的 JSON 字符串（存 Logs.extra TEXT） + }; + + // 日志级别 → 数值（大写兼容） + static const std::unordered_map& level_order() { + static const std::unordered_map 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 trace_id = std::nullopt, + std::optional extra = std::nullopt); + + void debug(const std::string& service, const std::string& message, + std::optional trace_id = std::nullopt, + std::optional 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 trace_id = std::nullopt, + std::optional 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 trace_id = std::nullopt, + std::optional 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 trace_id = std::nullopt, + std::optional 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 trace_id = std::nullopt, + std::optional 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&& 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 _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> _queue; +}; + +} // namespace qing diff --git a/liblogc/src/LogC.cpp b/liblogc/src/LogC.cpp new file mode 100644 index 0000000..2efd147 --- /dev/null +++ b/liblogc/src/LogC.cpp @@ -0,0 +1,141 @@ +/** LogC.cpp — 统一集中式日志客户端实现 + * + * 包含 _send_batch（pymsg → db execute INSERT）和 _utc_now_iso（ISO 8601）。 + * 编译为 libus_logc.so，运行时依赖 libus_msg.so。 + */ + +#include + +#include +#include +#include +#include +#include + +#include + +#include +#include + +namespace json = boost::json; + +namespace qing { + +// ---- 生命周期 --------------------------------------------------------------- + +LogC::LogC() + : _queue(std::make_unique>( + [this](std::vector&& 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&&) { + 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 lock(_endpoint_mtx); + _db_addr = addr; + _db_port = port; +} + +void LogC::set_service_name(const std::string& name) { + std::lock_guard 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 trace_id, + std::optional 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&& batch) { + std::string addr; + int port; + std::string client_name; + { + std::lock_guard 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( + 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(us)); + return buf; +} + +} // namespace qing diff --git a/pymsg/pymsg/__init__.py b/pymsg/pymsg/__init__.py index 5438a4f..e2bab70 100644 --- a/pymsg/pymsg/__init__.py +++ b/pymsg/pymsg/__init__.py @@ -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"] diff --git a/pymsg/pymsg/logc.py b/pymsg/pymsg/logc.py new file mode 100644 index 0000000..397a7a3 --- /dev/null +++ b/pymsg/pymsg/logc.py @@ -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) diff --git a/redirect80/log_client.py b/redirect80/log_client.py index dc3e905..cf26ca6 100644 --- a/redirect80/log_client.py +++ b/redirect80/log_client.py @@ -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") diff --git a/smtp/log_client.py b/smtp/log_client.py index 0067a14..c464cf2 100644 --- a/smtp/log_client.py +++ b/smtp/log_client.py @@ -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") diff --git a/smtp3/Makefile b/smtp3/Makefile index d729726..fd281e1 100644 --- a/smtp3/Makefile +++ b/smtp3/Makefile @@ -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 diff --git a/smtp3/inc/LogC.hpp b/smtp3/inc/LogC.hpp deleted file mode 100644 index 3e08bba..0000000 --- a/smtp3/inc/LogC.hpp +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include - -#include - -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>( - [this](std::vector&& 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&&) { - 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&& 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> queue_; - int fd_ = -1; // 长连接 fd（仅后台 flush 线程访问，无并发） -}; - -} // namespace smtp3 diff --git a/smtp3/src/main.cpp b/smtp3/src/main.cpp index 54a825b..bd59427 100644 --- a/smtp3/src/main.cpp +++ b/smtp3/src/main.cpp @@ -59,12 +59,12 @@ #include #include -#include "LogC.hpp" +#include 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 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 启动 diff --git a/static/log_client.py b/static/log_client.py index f1afa59..9a97a0a 100644 --- a/static/log_client.py +++ b/static/log_client.py @@ -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")