feat(core): confs2 合并入 core,内嵌 confs 配置服务 v0.1.0

- 新增 ConfsTask(ITask,与 HttpTask 同构):ConfigStore + INI/JSON 取值
  + get/list/reload/health 四路由,协议与 Python confs 逐字节兼容
- 新增 ConfsLogC 异步日志客户端(沿用 confs2 LogC,服务名 confs)
- main.cpp 新增 --confs <port> 选项与 CONFS_* 环境变量;不传则行为不变
- Makefile 链接 -lus_msg,版本 0.0.4 → 0.1.0
- _run.sh 启用 --confs 8010;_set.sh 自建 configs → ../confs/configs 链接
- init.txt 移除 confs2 自启动行(由 core 内嵌承担)
- 黑盒对比测试 16/16 通过(tests/blackbox/test_core_confs.py)
This commit is contained in:
qing committed 2026-08-09 15:36:32 +00:00
1 parent 1b282c6f1a
commit 59c5fe1feb
11 files changed
+947 -8

No files matched your search

+3
View File
@@ -130,3 +130,6 @@ go.sum
# VS
x64
.vs
# configs 符号链接(由 _set.sh 创建,指向 ../confs/configs/,内嵌 confs 服务使用)
configs
+6 -3
View File
@@ -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 $@
+12
View File
@@ -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 <port>`:启用内嵌 confs 配置服务(get/list/reload/health 四个路由)
- 不传 `--confs` 时行为与原版完全一致
- 配置文件目录默认为 `./configs`(_set.sh 会创建指向 `../confs/configs` 的符号链接)
- 环境变量:`CONFS_ADDR` / `CONFS_PORT` / `CONFS_WORKERS` / `CONFS_DIR`
## 如何使用并发机器运行程序
首先应该在模块中创建一个启动脚本,脚本中应该完成工作目录的切换、虚拟环境的激活,以及程序的启动。
+1 -1
View File
@@ -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
+12
View File
@@ -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
+132
View File
@@ -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 <us/Msg.hpp>
#include <us/AsyncQueue.hpp>
#include <boost/json.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <cstdlib>
#include <algorithm>
#include <cctype>
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<pmc::net::AsyncQueue<ConfsLogEntry>>(
[this](std::vector<ConfsLogEntry>&& 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<ConfsLogEntry>&&) {
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<ConfsLogEntry>&& 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<pmc::net::AsyncQueue<ConfsLogEntry>> queue_;
};
} // namespace confs
} // namespace qing
+63
View File
@@ -0,0 +1,63 @@
#pragma once
#include <memory>
#include <string>
#include <chrono>
#include <us/Thread.hpp>
#include <us/ITask.hpp>
#include <us/RpcServer.hpp>
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<ConfigStore> store; /* 配置仓库 */
std::unique_ptr<Thread> th; /* 可控线程指针 */
std::unique_ptr<pmc::net::RpcServer> 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);
};
}
-1
View File
@@ -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
+41 -3
View File
@@ -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<int>(), "kill a process") /* 杀死一个进程 */
/********* 子系统服务模式 ********/
("init,i", po::value<std::string>(), "self-init list") /* 输入自启动列表 */
("serve", po::value<int>(), "Run as http server"); /* 启用http伺服器 */
("serve", po::value<int>(), "Run as http server") /* 启用http伺服器 */
("confs", po::value<int>(), "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<int>();
auto confs = std::make_shared<ConfsTask> (addr, port, workers, dir);
confs->start();
taskPool.push_back(confs);
}
+557
View File
@@ -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 <us/RpcServer.hpp>
#include <boost/json.hpp>
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <ctime>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <mutex>
#include <optional>
#include <sstream>
#include <stdexcept>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#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<std::string> 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<std::chrono::system_clock::duration>(
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<std::chrono::microseconds>(
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<long long>(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<std::mutex> 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<char>(f)),
std::istreambuf_iterator<char>());
CacheEntry entry{content, mtime, content.size()};
cache_[name] = entry;
return entry; // 返回拷贝
}
// 列出配置目录全部文件元信息
std::vector<ConfigMeta> list() {
std::vector<ConfigMeta> out;
std::lock_guard<std::mutex> lock(mtx_);
std::string root_str = root_.string();
std::vector<std::string> 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<std::string> reload(const std::string* name) {
if (name != nullptr) {
validate_name(*name);
{
std::lock_guard<std::mutex> lock(mtx_);
resolve_locked(*name); // 锁内检查存在性,防止 TOCTOU
cache_.erase(*name);
}
get(*name); // 重新读取并缓存(get 内部自己加锁)
return {*name};
}
// 清空全部缓存
std::vector<std::string> dropped;
std::lock_guard<std::mutex> 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<std::mutex> 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<std::string, CacheEntry> 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<std::string> 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<std::string> 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<ConfigStore>(config_dir);
this->th = std::make_unique<Thread>();
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<pmc::net::RpcServer> (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<std::string>(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<std::string>(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<std::string>(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<std::chrono::seconds>(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}
};
});
}
}
+120
View File
@@ -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())