feat(gateway): 运行日志接入 db 集中式日志(新增 C++ 版 LogC,pymsg 推送 Logs 表)
- 新增 gateway/inc/LogC.hpp:异步批量日志客户端(内存队列+后台线程,50条/2秒 flush), 对齐 Python 版 log_client.py,fail-safe 设计(db 不可用静默丢弃) - main.cpp 全部日志输出迁移:wrapperHttp 双异常分支(>=500 ERROR / 其余 WARNING)、 启动/关闭、登录成功、注册成功、验证码邮件发送成功、代理白名单拒绝 - Makefile 增加 -pthread(LogC 使用 std::thread)
This commit is contained in:
1 parent
6a0551ab34
commit
f0cdaacd85
3 files changed
+277
-3
No files matched your search
+2
-2
@@ -4,8 +4,8 @@ INSTL_DIR := /usr/local/bin
|
||||
CPP := g++
|
||||
INCS := -I inc/
|
||||
MY_LIBS := -L/usr/lib -lus_http -lus_crypt
|
||||
LIBS := -lcurl -lssl -lsqlite3 -lz -lcrypto -lboost_json -lboost_program_options
|
||||
CPPFLAGS := --std=c++20 -pipe # -Wall -Werror
|
||||
LIBS := -pthread -lcurl -lssl -lsqlite3 -lz -lcrypto -lboost_json -lboost_program_options
|
||||
CPPFLAGS := --std=c++20 -pipe -pthread # -Wall -Werror
|
||||
BUILD_DIR = build
|
||||
$(shell mkdir -p $(BUILD_DIR))
|
||||
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
#pragma once
|
||||
|
||||
/** LogC — gateway 统一日志客户端(C++ 版,对应各 Python 服务的 log_client.py)
|
||||
*
|
||||
* 设计要点(与 Python 版对齐):
|
||||
* - 通过 pymsg TCP 调用 dbv2 execute 路由,逐条 INSERT 到 Logs 表
|
||||
* - 异步非阻塞:内存队列 + 后台线程批量推送,不拖慢业务
|
||||
* - fail-safe:db 服务不可用时静默丢弃(仅 stderr 提示),绝不影响业务
|
||||
* - 缓冲批量:队列满 50 条或超时 2 秒时自动 flush
|
||||
*
|
||||
* 用法:
|
||||
* #include "LogC.hpp"
|
||||
*
|
||||
* qing::LogC::instance().info("gateway", "启动完成");
|
||||
* qing::LogC::instance().error("gateway", "数据库连接失败",
|
||||
* std::nullopt, R"({"errno":2002})");
|
||||
*
|
||||
* // 进程退出前:
|
||||
* qing::LogC::instance().shutdown(); // 停止后台线程并 flush 剩余日志(幂等)
|
||||
*/
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstdio>
|
||||
#include <ctime>
|
||||
#include <deque>
|
||||
#include <iostream>
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <boost/json.hpp>
|
||||
|
||||
#include "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;
|
||||
}
|
||||
|
||||
// ---- 公开接口 --------------------------------------------------------
|
||||
|
||||
// 推送一条日志(非阻塞,立即返回)
|
||||
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) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_mtx);
|
||||
_queue.push_back(Entry{service, level, message,
|
||||
std::move(trace_id), std::move(extra)});
|
||||
}
|
||||
_cv.notify_one();
|
||||
}
|
||||
|
||||
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() {
|
||||
_force_flush.store(true);
|
||||
_cv.notify_all();
|
||||
for (int i = 0; i < 200 && _force_flush.load(); ++i) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
}
|
||||
}
|
||||
|
||||
// 停止后台线程并 flush 剩余日志(进程退出前调用,幂等)
|
||||
void shutdown() {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_mtx);
|
||||
if (_stop) return;
|
||||
_stop = true;
|
||||
}
|
||||
_cv.notify_all();
|
||||
if (_thread.joinable()) _thread.join();
|
||||
}
|
||||
|
||||
// ---- 配置(构造后、首次推送前可改)------------------------------------
|
||||
void set_endpoint(const std::string& addr, int port) {
|
||||
_db_addr = addr;
|
||||
_db_port = port;
|
||||
}
|
||||
|
||||
private:
|
||||
LogC() : _thread(&LogC::_flush_loop, this) {}
|
||||
~LogC() { shutdown(); }
|
||||
|
||||
LogC(const LogC&) = delete;
|
||||
LogC& operator=(const LogC&) = delete;
|
||||
|
||||
// 后台线程:积攒到 batch_size 或超时 flush_interval 时发送
|
||||
void _flush_loop() {
|
||||
std::vector<Entry> batch;
|
||||
auto last_flush = std::chrono::steady_clock::now();
|
||||
|
||||
for (;;) {
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(_mtx);
|
||||
// 最多 0.5s 轮询一次(对齐 Python 版 chk_timeout 上限)
|
||||
_cv.wait_for(lock, std::chrono::milliseconds(500),
|
||||
[&] { return _stop || !_queue.empty(); });
|
||||
while (!_queue.empty() && batch.size() < _BATCH_SIZE) {
|
||||
batch.push_back(std::move(_queue.front()));
|
||||
_queue.pop_front();
|
||||
}
|
||||
}
|
||||
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
const bool due = std::chrono::duration<double>(now - last_flush).count()
|
||||
>= _FLUSH_INTERVAL;
|
||||
if (!batch.empty() && (batch.size() >= _BATCH_SIZE || due
|
||||
|| _stop || _force_flush.load())) {
|
||||
_send(batch);
|
||||
batch.clear();
|
||||
last_flush = now;
|
||||
_force_flush.store(false);
|
||||
}
|
||||
|
||||
if (_stop) {
|
||||
// 退出前 flush 队列剩余
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(_mtx);
|
||||
while (!_queue.empty()) {
|
||||
batch.push_back(std::move(_queue.front()));
|
||||
_queue.pop_front();
|
||||
}
|
||||
}
|
||||
if (!batch.empty()) _send(batch);
|
||||
_force_flush.store(false);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表
|
||||
void _send(const std::vector<Entry>& batch) {
|
||||
const std::string now = _utc_now_iso(); // 批次共享同一时间戳(对齐 Python 版)
|
||||
try {
|
||||
pmc::net::MsgClient client(_db_addr, _db_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);
|
||||
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;
|
||||
static constexpr size_t _BATCH_SIZE = 50; // 队列满 50 条时 flush
|
||||
static constexpr double _FLUSH_INTERVAL = 2.0; // 超时 2 秒强制 flush
|
||||
static constexpr double _TIMEOUT = 5.0; // socket 超时(秒)
|
||||
|
||||
// ---- 状态 -------------------------------------------------------------
|
||||
std::deque<Entry> _queue;
|
||||
std::mutex _mtx;
|
||||
std::condition_variable _cv;
|
||||
bool _stop = false;
|
||||
std::atomic<bool> _force_flush{false};
|
||||
std::thread _thread;
|
||||
};
|
||||
|
||||
} // namespace qing
|
||||
+27
-1
@@ -20,6 +20,7 @@
|
||||
#include "LocalTime.hpp"
|
||||
#include "Conf.hpp"
|
||||
#include "DbC.hpp"
|
||||
#include "LogC.hpp"
|
||||
|
||||
using namespace pmc::net;
|
||||
using namespace qing;
|
||||
@@ -351,6 +352,9 @@ struct CaptchaEntry {
|
||||
static std::unordered_map<std::string, CaptchaEntry> _CAPTCHA_STORE;
|
||||
static std::mutex _CAPTCHA_MTX;
|
||||
|
||||
/* 集中式日志客户端:推送 gateway 运行日志到 db 模块 Logs 表(异步批量,fail-safe) */
|
||||
static inline LogC& logs() { return LogC::instance(); }
|
||||
|
||||
std::string generate_captcha_text() {
|
||||
static const char chars[] = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjkmnpqrstuvwxyz23456789";
|
||||
// thread_local:每个线程独立 rng 实例,避免多线程并发调用 dist(rng) 时的数据竞争
|
||||
@@ -535,10 +539,14 @@ int main(int argc, char **argv) {
|
||||
|
||||
|
||||
std::cout << "Gateway " << _VER << " started" << std::endl;
|
||||
logs().info("gateway", std::string("Gateway ") + _VER + " started");
|
||||
|
||||
/* 开始监听 */
|
||||
_SERVE->run();
|
||||
|
||||
/* 收到停止信号后的收尾:记录并 flush 剩余日志 */
|
||||
logs().info("gateway", "Gateway shutdown");
|
||||
logs().shutdown();
|
||||
}
|
||||
|
||||
//void db_init() {
|
||||
@@ -643,7 +651,14 @@ catch (const HttpException& e) {
|
||||
http::response<http::string_body> res;
|
||||
res.version(req.version());
|
||||
// 详细错误信息(e.what() 含源码路径/行号/C++ 模板类型)暂时以注释方式保留(同 handleLogin)
|
||||
std::cerr << "[handleRegister HttpException] " << e.what() << std::endl;
|
||||
std::cerr << "[wrapperHttp HttpException] " << e.what() << std::endl;
|
||||
// 推送集中式日志:5xx 服务端错误记 ERROR,其余(400/401/403/429 等业务错误)记 WARNING
|
||||
if (e.status() >= 500)
|
||||
logs().error("gateway", e.what(), std::nullopt,
|
||||
json::serialize(json::value{{"url", e.url()}, {"status", e.status()}}));
|
||||
else
|
||||
logs().warning("gateway", e.what(), std::nullopt,
|
||||
json::serialize(json::value{{"url", e.url()}, {"status", e.status()}}));
|
||||
json::value jv_r = {
|
||||
{"success", false},
|
||||
{"message", e.what()}
|
||||
@@ -667,6 +682,7 @@ catch (const std::exception& e) {
|
||||
// 透传到客户端会泄露 C++ 类型信息与内部路径(P0-3 遗漏点,已补全)
|
||||
// 排障:完整信息仍输出到服务端 stderr
|
||||
std::cerr << "[wrapperHttp std::exception] <" << ti.name() << "> " << e.what() << std::endl;
|
||||
logs().error("gateway", std::string("<") + ti.name() + "> " + e.what());
|
||||
json::value jv_r = {
|
||||
{"success", false},
|
||||
/* {"message", std::string("<") + ti.name() + "> " + e.what()}, */ // 泄露源码信息,已禁用
|
||||
@@ -879,6 +895,9 @@ http::response<http::string_body> handleLogin(
|
||||
res.set("Set-Cookie", ck);
|
||||
}
|
||||
|
||||
logs().info("gateway", "登录成功: " + username, std::nullopt,
|
||||
json::serialize(json::value{{"ip", client_ip}}));
|
||||
|
||||
res.result(http::status::ok);
|
||||
res.body() = json::serialize(jv_rr);
|
||||
res.prepare_payload();
|
||||
@@ -1037,6 +1056,9 @@ http::response<http::string_body> handleRegister(
|
||||
if (!success) THROW_EX_CUSTOM(HttpException, 2202, "Registe Failed", 400, "/register");
|
||||
auto jv_r = json::value{{"success", success}};
|
||||
|
||||
logs().info("gateway", "注册成功: " + username, std::nullopt,
|
||||
json::serialize(json::value{{"ip", client_ip}}));
|
||||
|
||||
res.result(http::status::ok);
|
||||
res.body() = json::serialize(jv_r);
|
||||
res.prepare_payload();
|
||||
@@ -1104,6 +1126,9 @@ http::response<http::string_body> handleSendEmailCode(
|
||||
if (!sent)
|
||||
THROW_EX_CUSTOM(HttpException, 3103, "Failed to send verification code", 500, "/register/send_code");
|
||||
|
||||
logs().info("gateway", "验证码邮件发送成功: " + email, std::nullopt,
|
||||
json::serialize(json::value{{"ip", client_ip}}));
|
||||
|
||||
auto jv_r = json::value{{"success", true}};
|
||||
res.result(http::status::ok);
|
||||
res.set(http::field::content_type, "application/json");
|
||||
@@ -1226,6 +1251,7 @@ http::response<http::string_body> handleProxy(
|
||||
}
|
||||
}
|
||||
if (!flag) {
|
||||
logs().warning("gateway", "代理目标不在白名单: " + t_url);
|
||||
throw std::runtime_error("403");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user