diff --git a/core/.gitignore b/core/.gitignore index 5e246e0..79115b1 100644 --- a/core/.gitignore +++ b/core/.gitignore @@ -130,3 +130,6 @@ go.sum # VS x64 .vs + +# configs 符号链接（由 _set.sh 创建，指向 ../confs/configs/，内嵌 confs 服务使用） +configs diff --git a/core/Makefile b/core/Makefile index 8f44c44..7e888e6 100644 --- a/core/Makefile +++ b/core/Makefile @@ -1,4 +1,4 @@ -VERSION := 0.0.4 +VERSION := 0.1.0 APP_NAME := nidus @@ -16,7 +16,7 @@ CPP:= g++ INCS:= -I include/ -MY_LIBS := -L/usr/lib -lus_th -lus_http -lus_httpclient -lcurl +MY_LIBS := -L/usr/lib -lus_th -lus_http -lus_httpclient -lus_msg -lcurl LIBS := -lpthread -lssl -lcrypto -lboost_program_options -lboost_json -lboost_thread -lboost_filesystem CPPFLAGS := --std=c++20 -O3 -pipe #-Wall -Werror @@ -26,7 +26,7 @@ $(shell mkdir -p $(BUILD_DIR)) #-------- all: ${APP_NAME}.out -${APP_NAME}.out: $(BUILD_DIR)/main.o $(BUILD_DIR)/PPool.o $(BUILD_DIR)/ProcessTask.o $(BUILD_DIR)/HttpTask.o $(BUILD_DIR)/Tttask.o $(BUILD_DIR)/pmc_mtd.o $(BUILD_DIR)/subsys_call.o +${APP_NAME}.out: $(BUILD_DIR)/main.o $(BUILD_DIR)/PPool.o $(BUILD_DIR)/ProcessTask.o $(BUILD_DIR)/HttpTask.o $(BUILD_DIR)/ConfsTask.o $(BUILD_DIR)/Tttask.o $(BUILD_DIR)/pmc_mtd.o $(BUILD_DIR)/subsys_call.o $(CPP) $^ $(MY_LIBS) $(LIBS) -o $@ $(BUILD_DIR)/main.o: src/main.cpp @@ -44,6 +44,9 @@ $(BUILD_DIR)/Tttask.o: src/main/Tttask.cpp $(BUILD_DIR)/HttpTask.o: src/main/HttpTask.cpp $(CPP) $(CPPFLAGS) -c $^ $(INCS) -o $@ +$(BUILD_DIR)/ConfsTask.o: src/main/ConfsTask.cpp + $(CPP) $(CPPFLAGS) -c $^ $(INCS) -o $@ + $(BUILD_DIR)/pmc_mtd.o: src/main/pmc_mtd.cpp $(CPP) $(CPPFLAGS) -c $^ $(INCS) -o $@ diff --git a/core/README.md b/core/README.md index 5e3f3b1..7d776a4 100644 --- a/core/README.md +++ b/core/README.md @@ -49,6 +49,18 @@ PMC在启动时生成一个AES密钥，然后用一个RSA公钥将其加密发 然后访问者与客户端之间的交流全部用此AES密钥加密 +## 内嵌 confs 配置服务（v0.1.0 起） +core 内置了统一配置服务（由 confs2 合并而来，pymsg RPC 协议，与 Python confs 完全兼容）： + +```bash +./nidus.out -i init.txt --serve 5200 --confs 8010 +``` + +- `--confs `：启用内嵌 confs 配置服务（get/list/reload/health 四个路由） +- 不传 `--confs` 时行为与原版完全一致 +- 配置文件目录默认为 `./configs`（_set.sh 会创建指向 `../confs/configs` 的符号链接） +- 环境变量：`CONFS_ADDR` / `CONFS_PORT` / `CONFS_WORKERS` / `CONFS_DIR` + ## 如何使用并发机器运行程序 首先应该在模块中创建一个启动脚本，脚本中应该完成工作目录的切换、虚拟环境的激活，以及程序的启动。 diff --git a/core/_run.sh b/core/_run.sh index e16b5f8..dde1669 100644 --- a/core/_run.sh +++ b/core/_run.sh @@ -1,4 +1,4 @@ #!/bin/bash cd "$(dirname "$0")" || exit export LD_LIBRARY_PATH=/lib:$LD_LIBRARY_PATH -./nidus.out -i init.txt --serve 5200 +./nidus.out -i init.txt --serve 5200 --confs 8010 diff --git a/core/_set.sh b/core/_set.sh index d8a7788..5a1049f 100644 --- a/core/_set.sh +++ b/core/_set.sh @@ -1,4 +1,16 @@ #!/bin/bash +set -euo pipefail cd "$(dirname "$0")" || exit + make clear && make $@ + +# 创建 configs 符号链接（指向 ../confs/configs/，内嵌 confs 服务配置文件零迁移） +if [ -L configs ] && [ ! -e configs ]; then + rm -f configs +fi +if [ ! -e configs ]; then + ln -s ../confs/configs configs + echo "[core] 已创建 configs → ../confs/configs 符号链接" +fi + make install diff --git a/core/include/main/ConfsLogC.hpp b/core/include/main/ConfsLogC.hpp new file mode 100644 index 0000000..cd3d8d5 --- /dev/null +++ b/core/include/main/ConfsLogC.hpp @@ -0,0 +1,132 @@ +/* + * 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/include/main/ConfsTask.hh b/core/include/main/ConfsTask.hh new file mode 100644 index 0000000..61b94ed --- /dev/null +++ b/core/include/main/ConfsTask.hh @@ -0,0 +1,63 @@ +#pragma once +#include +#include +#include +#include +#include +#include + +namespace qing { + +/* 配置仓库前置声明（实现位于 ConfsTask.cpp，保持头文件干净） */ +class ConfigStore; + +/* ---------------- + * 【内嵌 confs 配置服务任务】 + * ---------------- + * 将 confs2（C++20 统一配置服务，pymsg RpcServer）内嵌为 core 的一个 ITask， + * 与 HttpTask / Tttask 并列，由 taskPool 统一管理生命周期。 + * + * - 协议与 Python confs / standalone confs2 字节级兼容，消费方零改动 + * - 仅监听 127.0.0.1，路由 get/list/reload/health 不变 + * - 不传 --confs 参数时 core 不加载本任务，行为与原版完全一致 + */ +class ConfsTask: public ITask { +public: + + /* 构造内嵌配置服务任务 + * + * addr: 监听地址（默认 127.0.0.1） + * port: 监听端口（默认 8010） + * workers: RpcServer 工作线程数（默认 4） + * config_dir: 配置文件目录（默认 ./configs） */ + ConfsTask(std::string addr, int port, int workers, std::string config_dir); + + /* 析构需定义于 .cpp（unique_ptr 持有不完整类型 ConfigStore） */ + ~ConfsTask(); + + /* 任务是否在运行中 */ + bool isRunning() override; + + /* 启动任务（激活线程并等待 RpcServer 就绪） */ + void start() override; + + /* 终止任务（停止 RpcServer 并等待线程关闭） */ + void stop() override; + +private: + std::string addr; /* 监听地址 */ + int port; /* 监听端口 */ + int workers; /* 工作线程数 */ + std::unique_ptr store; /* 配置仓库 */ + std::unique_ptr th; /* 可控线程指针 */ + std::unique_ptr server; /* RPC 服务器指针 */ + std::chrono::steady_clock::time_point started_at; /* 启动时间（health uptime） */ + + /* 初始化线程，定义线程的行为（RpcServer） */ + void init_thread(); + + /* 注册 pymsg 路由（get/list/reload/health） */ + void register_routes(pmc::net::RpcServer& rpc); +}; + +} diff --git a/core/init.txt b/core/init.txt index 22f91aa..eff798f 100644 --- a/core/init.txt +++ b/core/init.txt @@ -1,5 +1,4 @@ bash /opt/micro_servers/db2/_run.sh -bash /opt/micro_servers/confs2/_run.sh bash /opt/micro_servers/ai_review/_run.sh bash /opt/micro_servers/dnfs2/_run.sh bash /opt/micro_servers/smtp/_run.sh diff --git a/core/src/main.cpp b/core/src/main.cpp index c38a7a9..b6ec8c0 100644 --- a/core/src/main.cpp +++ b/core/src/main.cpp @@ -32,6 +32,7 @@ #include "tt/Pipe.h" #include "main/Tttask.hh" #include "main/HttpTask.hh" +#include "main/ConfsTask.hh" #include "main/pmc_mtd.hh" #include "main/subsys_call.hh" @@ -39,7 +40,7 @@ #include "cn/中文化.hpp" /* 静态区 - 存放全局变量 */ -static const char *PMC_VERSION = "0.0.10"; /* 版本号 */ +static const char *PMC_VERSION = "0.1.0"; /* 版本号 */ static const size_t MSGLEN = 2048; /* 单条消息的长度 */ static const size_t MSGCNT = 100; /* 消息条数 */ @@ -52,8 +53,17 @@ namespace po = boost::program_options; void pmc_init(po::variables_map&); /* 初始化pmc并行机器 */ void pmc_serve(po::variables_map&); /* pmc的http伺服器模式 */ +void pmc_confs(po::variables_map&); /* pmc内嵌confs配置服务 */ void parse_data_from_mq(std::string&); /* 解析消息队列收到的消息 */ +/* 环境变量整数解析（失败时回退默认值，不终止进程） */ +static int env_or_default(const char* key, int fallback) { + auto* v = std::getenv(key); + if (!v) return fallback; + try { return std::stoi(v); } + catch (...) { return fallback; } +} + //------------------------------------------------------------- //------------------------------------------------------------- @@ -140,7 +150,8 @@ int main(int argc, char** argv) { ("kill", po::value(), "kill a process") /* 杀死一个进程 */ /********* 子系统服务模式 ********/ ("init,i", po::value(), "self-init list") /* 输入自启动列表 */ - ("serve", po::value(), "Run as http server"); /* 启用http伺服器 */ + ("serve", po::value(), "Run as http server") /* 启用http伺服器 */ + ("confs", po::value(), "Run embedded confs rpc server"); /* 启用内嵌confs配置服务 */ /* 参数变量映射关系 */ po::variables_map vm; @@ -210,7 +221,11 @@ int main(int argc, char** argv) { /* TODO: 抽象出来 */ if (vm.count("serve")) pmc_serve(vm); - + + /* 内嵌confs配置服务 */ + if (vm.count("confs")) + pmc_confs(vm); + /* pmc初始化函数 */ pmc_init(vm); @@ -259,6 +274,29 @@ void pmc_serve(po::variables_map& vm) } +/* 内嵌confs配置服务（pymsg RPC，只监听本地） + * + * 环境变量覆盖（CONFS_* 前缀，避免 core 环境变量污染子进程）： + * CONFS_ADDR 监听地址（默认 127.0.0.1） + * CONFS_PORT 监听端口（被 --confs 命令行参数覆盖） + * CONFS_WORKERS 工作线程数（默认 4） + * CONFS_DIR 配置文件目录（默认 ./configs） */ +void pmc_confs(po::variables_map& vm) +{ + std::string addr = "127.0.0.1"; + int workers = 4; + std::string dir = "./configs"; + if (auto* v = std::getenv("CONFS_ADDR")) addr = v; + if (auto* v = std::getenv("CONFS_DIR")) dir = v; + workers = env_or_default("CONFS_WORKERS", workers); + + int port = vm["confs"].as(); + + auto confs = std::make_shared (addr, port, workers, dir); + confs->start(); + taskPool.push_back(confs); +} + diff --git a/core/src/main/ConfsTask.cpp b/core/src/main/ConfsTask.cpp new file mode 100644 index 0000000..ce8f514 --- /dev/null +++ b/core/src/main/ConfsTask.cpp @@ -0,0 +1,557 @@ +/* + * ConfsTask.cpp — core 内嵌 confs 统一配置服务（pymsg RPC） + * + * 业务逻辑移植自 confs2/src/main.cpp（C++20 统一配置服务）， + * ConfigStore / INI 按键取值 / JSON 点路径 / 四个路由与原实现逐项对齐， + * 协议与 Python confs 字节级兼容，消费方零改动。 + * + * 与 standalone confs2 的差异： + * - 入口由 main() 改为 ConfsTask（ITask），生命周期由 core 的 taskPool 管理 + * - 引导配置由 core 的 main.cpp 解析（--confs / CONFS_* 环境变量）后注入 + */ + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "main/ConfsTask.hh" +#include "main/ConfsLogC.hpp" + +namespace json = boost::json; +namespace fs = std::filesystem; + +using qing::confs::ConfsLogC; + +namespace qing { + +// ═══════════════════════════════════════════════════════════════════════ +// 1. 常量与异常 +// ═══════════════════════════════════════════════════════════════════════ + +namespace { + +const std::unordered_set kAllowedExt = { + ".ini", ".conf", ".cfg", ".json", ".yaml", ".yml", ".toml", ".txt" +}; + +constexpr std::size_t kMaxConfigSize = 1 * 1024 * 1024; // 1 MiB + +class ConfigError : public std::runtime_error { +public: + using std::runtime_error::runtime_error; +}; +class ConfigNotFound : public ConfigError { +public: + using ConfigError::ConfigError; +}; +class ConfigNameInvalid : public ConfigError { +public: + using ConfigError::ConfigError; +}; +class KeyNotFound : public ConfigError { +public: + using ConfigError::ConfigError; +}; + +// ═══════════════════════════════════════════════════════════════════════ +// 2. 工具函数 +// ═══════════════════════════════════════════════════════════════════════ + +std::string get_extension(const std::string& name) { + auto dot = name.rfind('.'); + if (dot == std::string::npos) return ""; + std::string ext = name.substr(dot); + std::transform(ext.begin(), ext.end(), ext.begin(), + [](unsigned char c) { return std::tolower(c); }); + return ext; +} + +std::string iso_from_mtime(fs::file_time_type mtime) { + auto now_sys = std::chrono::system_clock::now(); + auto now_file = fs::file_time_type::clock::now(); + auto sys_tp = std::chrono::time_point_cast( + mtime - now_file + now_sys); + std::time_t t = std::chrono::system_clock::to_time_t(sys_tp); + + // 提取微秒部分（与 Python datetime.isoformat() 一致） + auto us = std::chrono::duration_cast( + sys_tp.time_since_epoch()).count() % 1000000; + + // 系统本地时间 + 时区偏移 + std::tm* lt = std::localtime(&t); + char buf[48]; + 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); // "+0000" or "+0800" + std::string tz = tz_raw; + if (tz.length() >= 5) + tz = tz.substr(0, 3) + ":" + tz.substr(3); // "+00:00" + + if (us == 0) + return std::string(buf) + tz; // Python: 微秒为0时省略 + char out[64]; + std::snprintf(out, sizeof(out), "%s.%06lld%s", + buf, static_cast(us), tz.c_str()); + return out; +} + +} // anonymous namespace + +// ═══════════════════════════════════════════════════════════════════════ +// 3. ConfigStore — 配置文件仓库（线程安全） +// ═══════════════════════════════════════════════════════════════════════ + +struct CacheEntry { + std::string content; + fs::file_time_type mtime; + std::size_t size; +}; + +struct ConfigMeta { + std::string name; + std::size_t size; + std::string mtime_iso; + bool cached; +}; + +class ConfigStore { +public: + explicit ConfigStore(const std::string& root) { + root_ = fs::weakly_canonical(fs::absolute(root)); + fs::create_directories(root_); + } + + std::string root_string() const { return root_.string(); } + + // 取配置全文（带缓存 + mtime 自动重载） + // 返回 CacheEntry 的拷贝（防缓存污染，与 Python 版 dict(entry) 语义一致） + CacheEntry get(const std::string& name) { + validate_name(name); // 锁外快速拒绝非法名 + std::lock_guard lock(mtx_); + auto path = resolve_locked(name); // 锁内: canonical + 前缀校验 + 存在性检查 + + // 锁内: stat → mtime 比对 → size 校验 → open → read，无 TOCTOU 窗口 + auto mtime = fs::last_write_time(path); + auto size = fs::file_size(path); + + auto it = cache_.find(name); + if (it != cache_.end() && it->second.mtime == mtime) + return it->second; // 缓存命中，返回拷贝 + + if (size > kMaxConfigSize) + throw ConfigError("配置文件过大: " + name); + + std::ifstream f(path); + if (!f.is_open()) + throw ConfigError("配置读取失败: " + name); + std::string content((std::istreambuf_iterator(f)), + std::istreambuf_iterator()); + + CacheEntry entry{content, mtime, content.size()}; + cache_[name] = entry; + return entry; // 返回拷贝 + } + + // 列出配置目录全部文件元信息 + std::vector list() { + std::vector out; + std::lock_guard lock(mtx_); + std::string root_str = root_.string(); + + std::vector names; + for (const auto& entry : fs::directory_iterator(root_)) { + auto name = entry.path().filename().string(); + if (kAllowedExt.find(get_extension(name)) == kAllowedExt.end()) + continue; + // 解析符号链接，逃逸的跳过 + auto path = fs::weakly_canonical(entry.path()); + std::string path_str = path.string(); + if (path_str.find(root_str) != 0 || path_str == root_str) + continue; + if (path_str.size() > root_str.size() && path_str[root_str.size()] != '/') + continue; + if (!fs::is_regular_file(path)) + continue; + names.push_back(name); + } + + std::sort(names.begin(), names.end()); + + for (const auto& name : names) { + auto path = fs::weakly_canonical(root_ / name); + try { + auto mtime = fs::last_write_time(path); + auto size = fs::file_size(path); + out.push_back({name, size, iso_from_mtime(mtime), + cache_.count(name) > 0}); + } catch (const std::exception&) { + continue; // 文件可能在遍历时被删除，跳过 + } + } + return out; + } + + // 重载指定配置 / 清空全部缓存 + std::vector reload(const std::string* name) { + if (name != nullptr) { + validate_name(*name); + { + std::lock_guard lock(mtx_); + resolve_locked(*name); // 锁内检查存在性，防止 TOCTOU + cache_.erase(*name); + } + get(*name); // 重新读取并缓存（get 内部自己加锁） + return {*name}; + } + // 清空全部缓存 + std::vector dropped; + std::lock_guard lock(mtx_); + for (const auto& [k, v] : cache_) + dropped.push_back(k); + cache_.clear(); + return dropped; + } + + std::size_t cached_count() const { + std::lock_guard lock(mtx_); + return cache_.size(); + } + +private: + // 扩展名白名单校验（纯字符串操作，不访问文件系统） + static void validate_name(const std::string& name) { + if (name.empty()) + throw ConfigNameInvalid("非法配置名: (空)"); + if (name.find('/') != std::string::npos || name.find('\\') != std::string::npos) + throw ConfigNameInvalid("非法配置名: '" + name + "'"); + auto ext = get_extension(name); + if (kAllowedExt.find(ext) == kAllowedExt.end()) + throw ConfigNameInvalid("不支持的配置类型: " + (ext.empty() ? "(无扩展名)" : ext)); + } + + // 持锁解析路径（canonical 解析符号链接 + 前缀校验 + 文件存在性） + fs::path resolve_locked(const std::string& name) { + auto path = fs::weakly_canonical(root_ / name); + std::string path_str = path.string(); + std::string root_str = root_.string(); + + // 校验路径在 root_ 之下 + if (path_str == root_str || + path_str.size() <= root_str.size() || + path_str[root_str.size()] != '/' || + path_str.find(root_str) != 0) + throw ConfigNameInvalid("非法配置名: '" + name + "'"); + + if (!fs::is_regular_file(path)) + throw ConfigNotFound("配置不存在: " + name); + + return path; + } + + fs::path root_; + std::unordered_map cache_; + mutable std::mutex mtx_; +}; + +namespace { + +// ═══════════════════════════════════════════════════════════════════════ +// 4. INI 按键取值 — 与 Python ConfigParser(interpolation=None) 行为对等 +// ═══════════════════════════════════════════════════════════════════════ +// +// 大小写策略（与 Python configparser 一致）： +// - Section 名：大小写敏感（[Server] ≠ [server]） +// - Option 名：大小写不敏感（port = Port = PORT） +// - 值：原样返回，不做任何转换 +// - 插值：禁用，%(key)s 原样返回 + +std::optional extract_ini_key( + const std::string& content, const std::string& key) +{ + auto dot = key.find('.'); + if (dot == std::string::npos || dot == 0 || dot == key.size() - 1) + return std::nullopt; + + std::string section = key.substr(0, dot); + std::string option = key.substr(dot + 1); + // option 大小写不敏感 → 转小写比较 + std::string option_lower = option; + std::transform(option_lower.begin(), option_lower.end(), option_lower.begin(), + [](unsigned char c) { return std::tolower(c); }); + + std::istringstream stream(content); + std::string line; + std::string current_section; + + while (std::getline(stream, line)) { + // 去 \r + if (!line.empty() && line.back() == '\r') + line.pop_back(); + + auto start = line.find_first_not_of(" \t"); + if (start == std::string::npos) continue; // 空行 + if (line[start] == ';' || line[start] == '#') continue; // 注释 + + if (line[start] == '[') { + auto end = line.find(']', start); + if (end != std::string::npos) + current_section = line.substr(start + 1, end - start - 1); + continue; + } + + auto eq = line.find('=', start); + if (eq == std::string::npos) continue; + + std::string key_name = line.substr(start, eq - start); + auto ke = key_name.find_last_not_of(" \t"); + if (ke != std::string::npos) key_name = key_name.substr(0, ke + 1); + + std::string key_lower = key_name; + std::transform(key_lower.begin(), key_lower.end(), key_lower.begin(), + [](unsigned char c) { return std::tolower(c); }); + + std::string value = line.substr(eq + 1); + auto vs = value.find_first_not_of(" \t"); + if (vs != std::string::npos) value = value.substr(vs); + auto ve = value.find_last_not_of(" \t\r"); + if (ve != std::string::npos) value = value.substr(0, ve + 1); + + if (current_section == section && key_lower == option_lower) + return value; + } + + return std::nullopt; +} + +// ═══════════════════════════════════════════════════════════════════════ +// 5. JSON 点路径取值 +// ═══════════════════════════════════════════════════════════════════════ + +json::value extract_json_key( + const std::string& content, const std::string& key) +{ + json::value node = json::parse(content); // 解析失败会抛异常 + + // 按点分割 + std::vector parts; + std::istringstream iss(key); + std::string part; + while (std::getline(iss, part, '.')) { + if (!part.empty()) parts.push_back(part); + } + + for (const auto& p : parts) { + if (node.is_object()) { + auto it = node.as_object().find(p); + if (it == node.as_object().end()) + throw KeyNotFound("键不存在: -> " + key); + node = it->value(); + } else if (node.is_array()) { + if (p.empty() || !std::all_of(p.begin(), p.end(), [](char c) { return std::isdigit(c); })) + throw KeyNotFound("数组下标必须是数字: " + p); + std::size_t idx = std::stoul(p); + if (idx >= node.as_array().size()) + throw KeyNotFound("数组下标越界: " + p); + node = node.as_array()[idx]; + } else { + throw KeyNotFound("无法继续遍历: " + p); + } + } + return node; +} + +} // anonymous namespace + +// ═══════════════════════════════════════════════════════════════════════ +// 6. ConfsTask — ITask 实现（生命周期与 HttpTask 同构） +// ═══════════════════════════════════════════════════════════════════════ + +ConfsTask::ConfsTask(std::string addr_, int port_, int workers_, std::string config_dir) + : addr(std::move(addr_)), port(port_), workers(workers_), + started_at(std::chrono::steady_clock::now()) +{ + store = std::make_unique(config_dir); + this->th = std::make_unique(); + init_thread(); +} + +ConfsTask::~ConfsTask() = default; + +bool ConfsTask::isRunning() { + auto stat = th->check(); + return !(stat == Fsm::Stat::STOP || stat == Fsm::Stat::SHUT); +} + +void ConfsTask::start() { + if (!this->isRunning()) { + th->Activate(); + th->WaitStart(); + } +} + +void ConfsTask::stop() { + if (this->server) this->server->stop(); + if (this->isRunning()) + th->WaitClose(); +} + +void ConfsTask::init_thread() { + + /* stop callback 静止事件 */ + this->th->set_stop( [](Thread& th) -> void { + th.suspend(); + }); + + /* wake callback 唤醒事件：创建 RpcServer 并阻塞运行 */ + this->th->set_wake( [this](Thread& th) -> void { + 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()); + }); + register_routes(*this->server); + th.run(); + + ConfsLogC::instance().info("confs", "开始监听 tcp://" + addr + ":" + + std::to_string(port)); + this->server->start(); /* 堵死线程，等待 stop() 解除 */ + }); + + /* loop callback 循环事件：服务器主动退出时停止线程防死循环 */ + this->th->set_loop( [](Thread& th) -> void { + th.stop(); + }); + + /* clean callback 清理事件 */ + this->th->set_clear( [this](Thread&) -> void { + this->server.reset(); + }); +} + +// ═══════════════════════════════════════════════════════════════════════ +// 7. 路由注册（get / list / reload / health，与 confs2 逐项对齐） +// ═══════════════════════════════════════════════════════════════════════ + +void ConfsTask::register_routes(pmc::net::RpcServer& rpc) { + + /* ── 路由: get ── */ + rpc.add_route("get", [this](const json::object& req) -> json::value { + auto name_it = req.find("name"); + if (name_it == req.end() || !name_it->value().is_string()) + throw ConfigError("缺少 name 字段"); + std::string name = json::value_to(name_it->value()); + + auto entry = store->get(name); + + // 无 key：返回全文 + auto key_it = req.find("key"); + if (key_it == req.end()) + return json::object{ + {"name", name}, + {"content", entry.content}, + {"size", (std::int64_t)entry.size}, + {"mtime", iso_from_mtime(entry.mtime)} + }; + + // 有 key：校验 + if (!key_it->value().is_string()) + throw ConfigError("key 字段必须是字符串"); + std::string key = json::value_to(key_it->value()); + if (key.empty()) + throw ConfigError("key 字段不能为空"); + + auto ext = get_extension(name); + + if (ext == ".ini" || ext == ".conf" || ext == ".cfg") { + auto val = extract_ini_key(entry.content, key); + if (!val) + throw KeyNotFound("键不存在: " + name + " -> " + key); + ConfsLogC::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); + return json::object{{"name", name}, {"key", key}, {"value", val}}; + } + + throw ConfigError(ext + " 类型不支持按键查询，请取全文后自行解析"); + }); + + /* ── 路由: list ── */ + rpc.add_route("list", [this](const json::object&) -> json::value { + auto configs = store->list(); + json::array arr; + for (const auto& c : configs) + arr.push_back(json::object{ + {"name", c.name}, + {"size", (std::int64_t)c.size}, + {"mtime", c.mtime_iso}, + {"cached", c.cached} + }); + return json::object{ + {"config_dir", store->root_string()}, + {"configs", arr} + }; + }); + + /* ── 路由: reload ── */ + rpc.add_route("reload", [this](const json::object& req) -> json::value { + auto it = req.find("name"); + + // 省略 name 或 null → 清空全部缓存 + if (it == req.end() || it->value().is_null()) { + auto names = store->reload(nullptr); + ConfsLogC::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}}; + } + + // 非字符串 → 报错 + if (!it->value().is_string()) + throw ConfigError("name 字段必须是字符串"); + + std::string name = json::value_to(it->value()); + if (name.empty()) + throw ConfigError("name 字段不能为空（要清空全部缓存请省略 name 字段）"); + + auto names = store->reload(&name); + ConfsLogC::instance().info("confs", "重载配置 " + names[0]); + json::array arr; + for (const auto& n : names) arr.push_back(json::value(n)); + return json::object{{"reloaded", arr}}; + }); + + /* ── 路由: health ── */ + rpc.add_route("health", [this](const json::object&) -> json::value { + auto now = std::chrono::steady_clock::now(); + auto uptime = std::chrono::duration_cast(now - started_at).count(); + return json::object{ + {"status", "ok"}, + {"service", "confs"}, + {"config_dir", store->root_string()}, + {"cached", (std::int64_t)store->cached_count()}, + {"uptime", uptime} + }; + }); +} + +} diff --git a/tests/blackbox/test_core_confs.py b/tests/blackbox/test_core_confs.py new file mode 100644 index 0000000..1228ec4 --- /dev/null +++ b/tests/blackbox/test_core_confs.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""core 内嵌 confs 黑盒测试 — 与 Python confs 行为对比 + +前置条件： + - Python confs 运行于 :8010（基线） + - core (nidus.out) 以 --confs 8012 启动 + +用法: python3 test_core_confs.py [core_port] [py_port] +""" +import os +import sys + +sys.path.insert(0, '/opt/micro_servers/pymsg') +from pymsg.rpc import RpcClient, RpcError + +CORE_PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8012 +PY_PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8010 +CONFIG_DIR = '/opt/micro_servers/confs/configs' + + +def call(port, route, payload=None): + return RpcClient('127.0.0.1', port, 'test').call(route, payload) + + +def call_err(port, route, payload=None): + """返回错误信息字符串；若未报错则返回 None""" + try: + call(port, route, payload) + return None + except RpcError as e: + return str(e) + + +passed = failed = 0 + + +def check(label, cond): + global passed, failed + if cond: + passed += 1 + print(f' [PASS] {label}') + else: + failed += 1 + print(f' [FAIL] {label}') + + +def main(): + print('========== 1. health ==========') + h = call(CORE_PORT, 'health') + print(' core:', h) + check('status=ok', h['status'] == 'ok') + check('service=confs', h['service'] == 'confs') + check('config_dir 指向 confs/configs', h['config_dir'].endswith('confs/configs')) + + print('========== 2. list ==========') + l2, l0 = call(CORE_PORT, 'list'), call(PY_PORT, 'list') + n2 = sorted(c['name'] for c in l2['configs']) + n0 = sorted(c['name'] for c in l0['configs']) + print(' core:', n2) + check('文件清单一致', n2 == n0) + check('size 一致', + {c['name']: c['size'] for c in l2['configs']} + == {c['name']: c['size'] for c in l0['configs']}) + + print('========== 3. get 全文 (db.ini) ==========') + g2 = call(CORE_PORT, 'get', {'name': 'db.ini'}) + g0 = call(PY_PORT, 'get', {'name': 'db.ini'}) + check('content 一致', g2['content'] == g0['content']) + check('size 一致', g2['size'] == g0['size']) + + print('========== 4. get INI 键 ==========') + import configparser + cp = configparser.ConfigParser(interpolation=None) + cp.read(os.path.join(CONFIG_DIR, 'db.ini')) + sec = cp.sections()[0] + opt = list(cp[sec].keys())[0] + key = f'{sec}.{opt}' + k2 = call(CORE_PORT, 'get', {'name': 'db.ini', 'key': key}) + k0 = call(PY_PORT, 'get', {'name': 'db.ini', 'key': key}) + print(f' key={key} core={k2["value"]!r} py={k0["value"]!r}') + check('INI 键值一致', k2['value'] == k0['value']) + + print('========== 5. 错误用例（两边都应报错且信息一致） ==========') + for label, payload in [('不存在的配置', {'name': 'nope.ini'}), + ('路径穿越', {'name': '../etc/x.ini'}), + ('非法类型', {'name': 'x.exe'}), + ('缺 name', None), + ('键不存在', {'name': 'db.ini', 'key': 'no.such'})]: + e2 = call_err(CORE_PORT, 'get', payload) + e0 = call_err(PY_PORT, 'get', payload) + print(f' {label}: core={e2!r} py={e0!r}') + check(f'{label} 错误一致', e2 is not None and e2 == e0) + + print('========== 6. reload ==========') + r2 = call(CORE_PORT, 'reload', {'name': 'db.ini'}) + r0 = call(PY_PORT, 'reload', {'name': 'db.ini'}) + check('reload 单个', r2 == r0 == {'reloaded': ['db.ini']}) + r2, r0 = call(CORE_PORT, 'reload'), call(PY_PORT, 'reload') + check('reload 全部返回集合一致', sorted(r2['reloaded']) == sorted(r0['reloaded'])) + + print('========== 7. INI option 大小写不敏感 ==========') + ku2 = call(CORE_PORT, 'get', {'name': 'db.ini', 'key': f'{sec}.{opt.upper()}'}) + check('大写 option 取值一致', ku2['value'] == k0['value']) + + print('========== 8. JSON 配置（若存在） ==========') + jsons = [f for f in os.listdir(CONFIG_DIR) if f.endswith('.json')] + if jsons: + j2 = call(CORE_PORT, 'get', {'name': jsons[0]}) + j0 = call(PY_PORT, 'get', {'name': jsons[0]}) + check(f'{jsons[0]} content 一致', j2['content'] == j0['content']) + else: + print(' (无 .json 配置，跳过)') + + print() + print(f'结果: {passed} 通过, {failed} 失败') + return 1 if failed else 0 + + +if __name__ == '__main__': + sys.exit(main())