diff --git a/db/log_client.py b/db/example_log_client.py similarity index 62% rename from db/log_client.py rename to db/example_log_client.py index d84d252..022b383 100644 --- a/db/log_client.py +++ b/db/example_log_client.py @@ -1,152 +1,185 @@ -""" -统一日志客户端 — 各服务共用。 - -设计要点: - - 使用 urllib.request(标准库,不引入 requests) - - 异步非阻塞:内存队列 + 后台线程批量推送,不拖慢业务 - - fail-safe:db 服务不可用时静默丢弃,绝不影响业务 - - 缓冲批量:队列满 50 条或超时 2 秒时自动 flush - -用法: - 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 threading -import queue -import time -import urllib.request -import urllib.error -import atexit - -# ── 配置 ────────────────────────────────────────────────────────────── - -DB_LOG_URL = 'http://127.0.0.1:8005/log/ingest' - -_BATCH_SIZE = 50 # 队列满 50 条时 flush -_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush -_TIMEOUT = 5 # HTTP 请求超时(秒) - - -class LogClient: - """异步批量日志推送客户端。""" - - def __init__(self, url=DB_LOG_URL, batch_size=_BATCH_SIZE, - flush_interval=_FLUSH_INTERVAL): - self._url = url - self._batch_size = batch_size - self._flush_interval = flush_interval - self._queue = queue.Queue() - self._stop = 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): - """推送一条日志(非阻塞,立即返回)。""" - self._queue.put({ - 'service': service, - 'level': level, - 'message': message, - 'trace_id': trace_id, - 'extra': extra, - }) - - 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 flush(self): - """手动 flush 队列中剩余的日志。""" - 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 时发送。""" - batch = [] - last_flush = time.monotonic() - - while not self._stop.is_set(): - timeout = max(0, self._flush_interval - (time.monotonic() - last_flush)) - try: - item = self._queue.get(timeout=timeout) - batch.append(item) - # 继续非阻塞地取,直到满 batch_size - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass # 超时,没有新日志 - - # 满了或超时了就发 - should_flush = ( - len(batch) >= self._batch_size or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush: - self._send(batch) - batch = [] - last_flush = time.monotonic() - - # 退出前 flush 剩余 - if batch: - self._send(batch) - - def _send(self, batch): - """发送一批日志到 db 服务。失败时静默丢弃。""" - data = json.dumps({'logs': batch}).encode('utf-8') - req = urllib.request.Request( - self._url, data=data, - headers={'Content-Type': 'application/json'}, - method='POST' - ) - try: - with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: - resp.read() # 消费响应体 - except (urllib.error.URLError, urllib.error.HTTPError, OSError): - # fail-safe:日志服务不可用,静默丢弃,不影响业务 - pass - except Exception: - pass - - def _shutdown(self): - """进程退出时 flush 剩余日志。""" - self._stop.set() - self.flush() - - -# ── 全局单例 ────────────────────────────────────────────────────────── - -logs = LogClient() +""" +统一日志客户端 SDK — 各服务复制即用。 + +设计要点: + - 通过 pymsg TCP 调用 dbv2 execute 路由,逐条 INSERT 到 Logs 表 + - 异步非阻塞:内存队列 + 后台线程批量推送,不拖慢业务 + - fail-safe:db 服务不可用时静默丢弃,绝不影响业务 + - 缓冲批量:队列满 50 条或超时 2 秒时自动 flush + +复制到目标服务目录后,按需修改 service_name 即可:: + + 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 MsgClient # noqa: E402 + +# ── 配置 ────────────────────────────────────────────────────────────── + +_DBV2_ADDR = '127.0.0.1' +_DBV2_PORT = 8005 + +_BATCH_SIZE = 50 # 队列满 50 条时 flush +_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush +_TIMEOUT = 5 # socket 超时(秒) + + +class LogClient: + """异步批量日志推送客户端。 + + 通过 MsgClient(pymsg TCP 协议)逐条 INSERT 到 dbv2 的 Logs 表。 + service_name 用于在 dbv2 连接中标识来源,各服务复制时自行修改。 + """ + + def __init__(self, service_name="example", + 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() + self._stop = 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): + """推送一条日志(非阻塞,立即返回)。""" + self._queue.put({ + 'service': service, + 'level': level, + 'message': message, + 'trace_id': trace_id, + 'extra': extra, + }) + + 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 flush(self): + """手动 flush 队列中剩余的日志。""" + 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 时发送。""" + batch = [] + last_flush = time.monotonic() + + while not self._stop.is_set(): + timeout = max(0, self._flush_interval - (time.monotonic() - last_flush)) + try: + item = self._queue.get(timeout=timeout) + batch.append(item) + # 继续非阻塞地取,直到满 batch_size + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + # 超时无新日志:如果 batch 中已有积攒的条目,强制 flush + if batch: + self._send(batch) + batch = [] + last_flush = time.monotonic() + continue + + # 满了或超时了就发 + should_flush = ( + len(batch) >= self._batch_size or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush: + self._send(batch) + batch = [] + last_flush = time.monotonic() + + # 退出前 flush 剩余 + if batch: + self._send(batch) + + def _send(self, batch): + """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" + now = datetime.now(timezone.utc).isoformat() + try: + client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") + for entry in batch: + extra_val = entry.get("extra") + payload = { + "_route": "execute", + "action": "insert", + "table": "Logs", + "values": { + "ts": now, + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": json.dumps(extra_val, ensure_ascii=False) + if extra_val is not None else None, + }, + } + resp = client.call(json.dumps(payload), timeout=_TIMEOUT) + resp_obj = json.loads(resp) + if not resp_obj.get("success"): + import sys + print(f"[log_client] INSERT 失败: {resp_obj.get('error')}", + file=sys.stderr) + except Exception as e: + import sys + print(f"[log_client] 发送失败: {e}", file=sys.stderr) + + def _shutdown(self): + """进程退出时 flush 剩余日志。""" + self._stop.set() + self.flush() + + +# ── 全局单例 ────────────────────────────────────────────────────────── +# 各服务复制时,把 "example" 改为自己的服务名即可。 +# 例如 dnfs/log_client.py → logs = LogClient(service_name="dnfs") +# 例如 smtp/log_client.py → logs = LogClient(service_name="smtp") + +logs = LogClient() diff --git a/gateway/inc/DbC.hpp b/gateway/inc/DbC.hpp index 9261625..7ca99c9 100644 --- a/gateway/inc/DbC.hpp +++ b/gateway/inc/DbC.hpp @@ -1,10 +1,13 @@ #pragma once #include +#include "Msg.hpp" #include namespace qing { namespace json = boost::json; class DbC { public: + // 【保留】旧版 HTTP 接口:通过 HttpClient POST 到 dbv2 HTTP /execute 端点。 + // 保留用于兼容性参考,新代码优先使用 call_msg()(pymsg TCP 协议,更可靠)。 static json::value call_http(json::value& params) { auto client = pmc::net::HttpClient(); @@ -16,6 +19,21 @@ namespace qing { }); return json::parse(res); } + + // 新版 pymsg 接口:通过 MsgClient(TCP 长度前缀帧协议)调用 dbv2。 + // 自动添加 _route = "execute",返回 dbv2 pymsg 完整 JSON 响应 + // {"success":..., "data":[...]}。 + // 响应格式与 call_http 兼容,调用方可直接替换。 + static json::value call_msg(const json::value& params, + double timeout = 5.0) { + json::value jv_route = params; + jv_route.as_object()["_route"] = "execute"; + pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); + std::string resp = client.call( + json::serialize(jv_route), timeout); + return json::parse(resp); + } + private: static constexpr const char* _DBURL = "http://127.0.0.1:8005/execute"; }; diff --git a/gateway/inc/Msg.hpp b/gateway/inc/Msg.hpp new file mode 100644 index 0000000..fd7a956 --- /dev/null +++ b/gateway/inc/Msg.hpp @@ -0,0 +1,334 @@ +#pragma once + +/** pymsg C++ 实现 — Length-prefixed variable-length socket message framing. + * + * Protocol: [4-byte big-endian uint32_t length][UTF-8 payload] + * + * 对应 Python 版 pymsg/pymsg/__init__.py,提供: + * - pmsg_send() / pmsg_recv() 底层协议收发函数 + * - MsgClient 短连接 send / call / 长连接 listen 客户端 + * + * Usage: + * #include "Msg.hpp" + * + * pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); + * std::string resp = client.call(R"({"sql":"SELECT 1"})", 5.0); + * client.send("hello"); + */ + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +// macOS 没有 MSG_NOSIGNAL,用 0 替代(内核不会发 SIGPIPE) +#ifndef MSG_NOSIGNAL +#define MSG_NOSIGNAL 0 +#endif + +namespace pmc { +namespace net { + +// ============================================================================ +// Exceptions +// ============================================================================ + +class MsgException : public std::runtime_error { +public: + explicit MsgException(const std::string &msg) : std::runtime_error(msg) {} +}; + +class MsgTimeoutException : public MsgException { +public: + explicit MsgTimeoutException(const std::string &msg) : MsgException(msg) {} +}; + +class MsgConnectionException : public MsgException { +public: + explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {} +}; + +// ============================================================================ +// Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload +// ============================================================================ + +namespace detail { + +constexpr size_t kHeaderSize = 4; + +// ---- 4-byte big-endian 编码 / 解码 ---------------------------------------- + +inline void writeUint32BE(char *buf, uint32_t val) noexcept { + buf[0] = static_cast((val >> 24) & 0xFFu); + buf[1] = static_cast((val >> 16) & 0xFFu); + buf[2] = static_cast((val >> 8) & 0xFFu); + buf[3] = static_cast(val & 0xFFu); +} + +inline uint32_t readUint32BE(const char *buf) noexcept { + return (static_cast(static_cast(buf[0])) << 24) | + (static_cast(static_cast(buf[1])) << 16) | + (static_cast(static_cast(buf[2])) << 8) | + (static_cast(static_cast(buf[3]))); +} + +// ---- socket 超时设置 ------------------------------------------------------ + +inline void setSockTimeout(int sockfd, double seconds) { + struct timeval tv {}; + tv.tv_sec = static_cast(seconds); + tv.tv_usec = + static_cast((seconds - static_cast(tv.tv_sec)) * 1'000'000.0); + ::setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + ::setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); +} + +// ---- 精确读取 n 字节 ------------------------------------------------------ + +inline void recvExact(int sockfd, char *buf, size_t n) { + size_t received = 0; + while (received < n) { + ssize_t r = ::recv(sockfd, buf + received, n - received, 0); + if (r == 0) { + throw MsgConnectionException( + "connection closed: expected " + std::to_string(n) + + " bytes, got " + std::to_string(received)); + } + if (r < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + throw MsgTimeoutException("recv timed out"); + } + throw MsgConnectionException( + "recv error: " + std::string(::strerror(errno))); + } + received += static_cast(r); + } +} + +// ---- 完整发送 ------------------------------------------------------------- + +inline void sendAll(int sockfd, const char *data, size_t len) { + size_t sent = 0; + while (sent < len) { + ssize_t w = + ::send(sockfd, data + sent, len - sent, MSG_NOSIGNAL); + if (w < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + throw MsgTimeoutException("send timed out"); + } + throw MsgConnectionException( + "send error: " + std::string(::strerror(errno))); + } + sent += static_cast(w); + } +} + +// ---- 短连接 helper:connect → action → close ------------------------------ + +inline int openConnection(const std::string& addr, int port, double timeout) { + int fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + throw MsgConnectionException( + "socket() failed: " + std::string(::strerror(errno))); + } + + struct sockaddr_in serv_addr {}; + serv_addr.sin_family = AF_INET; + serv_addr.sin_port = htons(static_cast(port)); + if (::inet_pton(AF_INET, addr.c_str(), &serv_addr.sin_addr) <= 0) { + ::close(fd); + throw MsgConnectionException("invalid address: " + addr); + } + + setSockTimeout(fd, timeout); + + if (::connect(fd, reinterpret_cast(&serv_addr), + sizeof(serv_addr)) < 0) { + ::close(fd); + throw MsgConnectionException( + "could not connect to " + addr + ":" + std::to_string(port) + + " - " + ::strerror(errno)); + } + + return fd; +} + +} // namespace detail + +// ============================================================================ +// 底层协议函数:pmsg_send / pmsg_recv +// ============================================================================ + +/// 以 pymsg 协议发送一条字符串消息(长度前缀 + UTF-8 载荷)。 +inline void pmsg_send(int sockfd, const std::string &data) { + size_t payload_size = data.size(); + if (payload_size > 0xFFFFFFFFu) { + throw MsgException("payload too large: " + std::to_string(payload_size)); + } + char header[detail::kHeaderSize]; + detail::writeUint32BE(header, static_cast(payload_size)); + detail::sendAll(sockfd, header, detail::kHeaderSize); + detail::sendAll(sockfd, data.c_str(), payload_size); +} + +/// 以 pymsg 协议接收一条字符串消息。 +/// @throws MsgConnectionException 连接在收全消息前断开。 +/// @throws MsgTimeoutException 等待超时。 +inline std::string pmsg_recv(int sockfd) { + char header[detail::kHeaderSize]; + detail::recvExact(sockfd, header, detail::kHeaderSize); + uint32_t length = detail::readUint32BE(header); + if (length > 64 * 1024 * 1024) { // 安全上限 64 MiB + throw MsgException("message too large: " + std::to_string(length)); + } + std::string payload(static_cast(length), '\0'); + detail::recvExact(sockfd, payload.data(), length); + return payload; +} + +// ============================================================================ +// MsgClient +// ============================================================================ + +/// 基于 pymsg 协议的 TCP 客户端。 +/// +/// 支持三种操作模式: +/// - send() 短连接发送一条消息(不等待回复),消息带 name 前缀。 +/// - call() 短连接发送 + 阻塞等待回复(RPC 模式),原始消息不添加前缀。 +/// - listen() 基于 connect() 维持的长连接,持续接收消息并回调。 +class MsgClient { +public: + MsgClient(const std::string &addr, int port, const std::string &name) + : _addr(addr), _port(port), _name(name), _sockfd(-1), _running(false) {} + + ~MsgClient() { stop(); } + + // ---- 连接 --------------------------------------------------------------- + + /// 创建 TCP 套接字并连接到服务器(长连接,供 listen() 使用)。 + void connect() { + _sockfd = detail::openConnection(_addr, _port, /*timeout=*/10.0); + } + + // ---- 发送单条消息(短连接,带名字前缀)---------------------------------- + + /// 短连接发送一条消息到服务器,消息格式为 ``"name: msg"``。 + /// 不等待回复。短连接(每次创建新 socket,发完即关)。 + void send(const std::string &msg) { + int fd = detail::openConnection(_addr, _port, /*timeout=*/3.0); + try { + pmsg_send(fd, _name + ": " + msg); + ::close(fd); + } catch (...) { + ::close(fd); + throw; + } + } + + // ---- 请求-应答(短连接,RPC 模式)--------------------------------------- + + /// 短连接发送一条消息,阻塞等待并返回一条回复。 + /// + /// 不会给消息添加 ``name:`` 前缀(适用于 RPC/API 调用)。 + /// + /// @param msg 要发送的原始字符串。 + /// @param timeout 超时秒数,默认 10.0。 + /// @returns 服务器返回的字符串。 + std::string call(const std::string &msg, double timeout = 10.0) { + int fd = detail::openConnection(_addr, _port, timeout); + try { + pmsg_send(fd, msg); + std::string resp = pmsg_recv(fd); + ::close(fd); + return resp; + } catch (...) { + ::close(fd); + throw; + } + } + + // ---- 持续监听(长连接,阻塞回调模式)----------------------------------- + + /// 在长连接上持续接收消息并通过回调处理。 + /// + /// 阻塞当前线程直到 stop() 被调用或连接断开。 + /// 必须先调用 connect()。 + /// + /// @param on_message 回调 ``void(const std::string &msg)``。 + void listen(std::function on_message) { + if (_sockfd < 0) { + throw MsgException("must call connect() before listen()"); + } + + _running = true; + while (_running) { + fd_set readfds; + FD_ZERO(&readfds); + FD_SET(_sockfd, &readfds); + + struct timeval tv {}; + tv.tv_sec = 0; + tv.tv_usec = 100'000; // 100 ms 轮询间隔 + + int ready = + ::select(_sockfd + 1, &readfds, nullptr, nullptr, &tv); + if (ready < 0) { + if (errno == EINTR) continue; + break; // EBADF or other fatal error + } + if (ready == 0) continue; // timeout, loop around + + try { + std::string text = pmsg_recv(_sockfd); + on_message(std::move(text)); + } catch (const MsgConnectionException &) { + break; // 连接断开,退出循环 + } catch (const MsgTimeoutException &) { + continue; // 读超时但连接仍在,继续等待 + } + } + + if (_sockfd >= 0) { + ::close(_sockfd); + _sockfd = -1; + } + } + + // ---- 停止 --------------------------------------------------------------- + + /// 设置停止标志并关闭套接字。可从任意线程调用。 + void stop() noexcept { + _running = false; + if (_sockfd >= 0) { + ::close(_sockfd); + _sockfd = -1; + } + } + + // ---- 访问器 ------------------------------------------------------------- + + const std::string &addr() const noexcept { return _addr; } + int port() const noexcept { return _port; } + const std::string &name() const noexcept { return _name; } + +private: + std::string _addr; + int _port; + std::string _name; + int _sockfd; + bool _running; +}; + +} // namespace net +} // namespace pmc diff --git a/gateway/src/main.cpp b/gateway/src/main.cpp index bd9218a..fec25f1 100644 --- a/gateway/src/main.cpp +++ b/gateway/src/main.cpp @@ -700,7 +700,9 @@ http::response handleLogin( // stored_hash = std::string(reinterpret_cast(stmt.read_text(0))); //} - // 新版调用db模块 + // 新版调用db模块(通过 MsgClient pymsg TCP 协议) + // 【保留】旧版 HTTP 调用方式: + // auto jv_db_r = DbC::call_http(jv_select); json::value jv_select = { {"action", "select"}, {"table", "Users"}, @@ -708,7 +710,7 @@ http::response handleLogin( {"username", username}, }} }; - auto jv_db_r = DbC::call_http(jv_select); + auto jv_db_r = DbC::call_msg(jv_select); auto obj_r = jv_db_r.as_object(); auto success = obj_r["success"].as_bool(); if (!success) THROW_EX_CUSTOM(HttpException, 2102, "Login (Update Step) Failed", 500, "/login"); @@ -853,7 +855,9 @@ http::response handleRegister( //stmt.bind_text(2, password_hash); //stmt.step(); // - // 新版调用db模块 + // 新版调用db模块(通过 MsgClient pymsg TCP 协议) + // 【保留】旧版 HTTP 调用方式: + // auto jv_result = DbC::call_http(jv_insert); json::value jv_insert = { {"action", "insert"}, {"table", "Users"}, @@ -862,7 +866,7 @@ http::response handleRegister( {"password_hash", password_hash} }} }; - auto jv_result = DbC::call_http(jv_insert); + auto jv_result = DbC::call_msg(jv_insert); auto obj_r = jv_result.as_object(); auto success = obj_r["success"].as_bool(); if (!success) THROW_EX_CUSTOM(HttpException, 2202, "Registe Failed", 400, "/register");