diff --git a/gateway/Makefile b/gateway/Makefile index a6c49ef..e144517 100644 --- a/gateway/Makefile +++ b/gateway/Makefile @@ -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)) diff --git a/gateway/inc/LogC.hpp b/gateway/inc/LogC.hpp new file mode 100644 index 0000000..a685d9e --- /dev/null +++ b/gateway/inc/LogC.hpp @@ -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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#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 trace_id; + std::optional 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 trace_id = std::nullopt, + std::optional extra = std::nullopt) { + { + std::lock_guard 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 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() { + _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 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 batch; + auto last_flush = std::chrono::steady_clock::now(); + + for (;;) { + { + std::unique_lock 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(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 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& 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( + 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; + 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 _queue; + std::mutex _mtx; + std::condition_variable _cv; + bool _stop = false; + std::atomic _force_flush{false}; + std::thread _thread; +}; + +} // namespace qing diff --git a/gateway/src/main.cpp b/gateway/src/main.cpp index 9bb713e..e6b1989 100644 --- a/gateway/src/main.cpp +++ b/gateway/src/main.cpp @@ -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 _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 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 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 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 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 handleProxy( } } if (!flag) { + logs().warning("gateway", "代理目标不在白名单: " + t_url); throw std::runtime_error("403"); } }