Skip to content

Commit da7f2a5

Browse files
JackarainJackarain
committed
launcher 将用户累计上传/下载持久化到服务端并提供重置接口
1 parent 2d34bfe commit da7f2a5

11 files changed

Lines changed: 355 additions & 427 deletions

File tree

‎apps/launcher/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ launcher 启动 `proxy_server` 时通过 `--launcher ws://host:port/rpc?instance
2929
`ssl_prefer_server_ciphers`），其余选项立即生效；WebUI 另按选项表的 `restart_only`
3030
标记（含 `tun` / `tun_name` / `tun_mtu` / `proxy_domains` / `proxy_cidr`）作静态提示。
3131
- **用户管理**：新增/删除用户、修改密码、设置限速、流量配额与连接数限制，配置与运行期双向同步。
32-
- **用量续接**：用户累计流量在 launcher 侧持久化，实例重启后经 `set_user_usage` 续接配额计数。
32+
- **用量续接与统计**：按用户持久化累计上传/下载（实例上报的会话级计数重启归零，launcher 折算增量后累积）与含续接基线的配额总量；实例重启后经 `set_user_usage` 续接配额计数。WebUI 可直接重置单个用户或全部用户的累计上传/下载（配额总量不变）。
3333
- **状态监控**：实例经控制通道上报状态，WebUI 展示在线状态、PID、连接数、上下行速率与连接明细。
3434
- **日志采集**：每个实例保留最近 2000 行（环形缓冲），支持按序号增量拉取。
3535
- **崩溃自动重启**：实例意外退出后自动拉起；60 秒内连续崩溃超过 3 次则停止重启并告警。

‎apps/launcher/http_server.cpp‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1249,6 +1249,27 @@ net::awaitable<response> http_server::route(const http::request<http::string_bod
12491249
co_return make_error(http::status::method_not_allowed, "method not allowed");
12501250
}
12511251

1252+
// 重置用户累计上传/下载；请求体 {"user":"..."} 重置单个用户，
1253+
// 缺省或空 user 重置该实例全部用户。
1254+
if (action == "usage" && parts.size() >= 3 && parts[2] == "reset") {
1255+
if (req.method() != http::verb::post)
1256+
co_return make_error(http::status::method_not_allowed, "method not allowed");
1257+
std::string user;
1258+
if (!req.body().empty()) {
1259+
boost::system::error_code ec;
1260+
auto jv = json::parse(req.body(), ec);
1261+
if (!ec && jv.is_object()) {
1262+
if (auto u = jv.as_object().if_contains("user"); u && u->is_string())
1263+
user = std::string(u->as_string());
1264+
}
1265+
}
1266+
json::value out;
1267+
std::string err;
1268+
if (!m_mgr_->reset_usage(id, user, out, err))
1269+
co_return make_error(http::status::not_found, err);
1270+
co_return make_json_response(http::status::ok, out);
1271+
}
1272+
12521273
co_return make_error(http::status::not_found, "not found");
12531274
}
12541275

‎apps/launcher/manager.cpp‎

Lines changed: 133 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,45 @@ std::string as_string(const json::value& v)
7474
return {};
7575
}
7676

77+
// 读取对象的数值字段（缺失/类型不符返回 0）。
78+
std::int64_t obj_num(const json::object& o, const char* key)
79+
{
80+
auto it = o.find(key);
81+
return it == o.end() ? 0 : as_int64(it->value());
82+
}
83+
84+
// 用量记录 -> JSON（对外视图，不含会话级折算用的 last_* 字段）。
85+
json::object usage_view(const std::map<std::string, usage_record>& m)
86+
{
87+
json::object out;
88+
for (const auto& [user, rec] : m) {
89+
json::object r;
90+
r["total"] = rec.total;
91+
r["rx"] = rec.rx;
92+
r["tx"] = rec.tx;
93+
out[user] = std::move(r);
94+
}
95+
return out;
96+
}
97+
98+
// 从持久化 JSON 解析单个用户用量；兼容旧版单值（合并总量）格式。
99+
usage_record parse_usage_record(const json::value& v)
100+
{
101+
usage_record rec;
102+
if (v.is_object()) {
103+
const auto& o = v.as_object();
104+
rec.total = obj_num(o, "total");
105+
rec.rx = obj_num(o, "rx");
106+
rec.tx = obj_num(o, "tx");
107+
rec.last_rx = obj_num(o, "last_rx");
108+
rec.last_tx = obj_num(o, "last_tx");
109+
} else {
110+
// 旧版仅持久化合并总量，无法拆分方向：计入配额基线，展示口径从零起算。
111+
rec.total = as_int64(v);
112+
}
113+
return rec;
114+
}
115+
77116
// 随机 hex 字符串（crypto 强度，跨平台：OpenSSL RAND_bytes）。
78117
std::string random_hex(std::size_t bytes)
79118
{
@@ -397,8 +436,10 @@ bool manager::load_from_disk()
397436
in->config_ = c->as_object();
398437
if (in->config_.empty())
399438
in->config_ = json::object();
400-
if (auto u = obj.if_contains("user_usage"); u && u->is_object())
401-
in->user_usage_ = u->as_object();
439+
if (auto u = obj.if_contains("user_usage"); u && u->is_object()) {
440+
for (const auto& [user, v] : u->as_object())
441+
in->user_usage_[std::string(user)] = parse_usage_record(v);
442+
}
402443
if (auto t = obj.if_contains("created_at"); t && t->is_string()) {
403444
time_point tp;
404445
if (rfc3339_parse(std::string(t->as_string()), tp))
@@ -430,25 +471,42 @@ std::vector<std::string> manager::ids()
430471

431472
bool manager::save()
432473
{
433-
std::vector<std::pair<instance_ptr, json::object>> list;
474+
struct save_item
475+
{
476+
instance_ptr in;
477+
json::object cfg;
478+
std::map<std::string, usage_record> usage;
479+
};
480+
std::vector<save_item> list;
434481
{
435482
std::lock_guard<std::mutex> lock(m_mu_);
436483
list.reserve(m_instances_.size());
437484
for (const auto& [_, in] : m_instances_) {
438485
// 深拷贝 map：锁外序列化期间其他线程可能在锁内就地修改 config_/user_usage_。
439-
list.emplace_back(in, json::object(in->config_));
486+
list.push_back(save_item{ in, json::object(in->config_), in->user_usage_ });
440487
}
441488
}
442489
json::array arr;
443-
for (auto& [in, cfg] : list) {
490+
for (auto& it : list) {
491+
auto& in = it.in;
492+
json::object usage;
493+
for (const auto& [user, rec] : it.usage) {
494+
json::object r;
495+
r["total"] = rec.total;
496+
r["rx"] = rec.rx;
497+
r["tx"] = rec.tx;
498+
r["last_rx"] = rec.last_rx;
499+
r["last_tx"] = rec.last_tx;
500+
usage[user] = std::move(r);
501+
}
444502
json::object item;
445503
item["id"] = in->id_;
446504
item["name"] = in->name_;
447-
item["config"] = std::move(cfg);
505+
item["config"] = std::move(it.cfg);
448506
item["autostart"] = in->autostart_;
449507
item["token"] = in->token_;
450508
item["created_at"] = rfc3339_format(in->created_at_);
451-
item["user_usage"] = json::object(in->user_usage_);
509+
item["user_usage"] = std::move(usage);
452510
arr.emplace_back(std::move(item));
453511
}
454512

@@ -1255,6 +1313,48 @@ bool manager::status_view(const std::string& id, json::value& out)
12551313
o["pid"] = in->pid();
12561314
o["last_seen"] = rfc3339_format(in->last_seen_);
12571315
o["report"] = in->last_report_.is_object() ? in->last_report_ : json::value(json::object_kind);
1316+
// 各用户累计上下行（WebUI 展示口径），与实时报告合并返回。
1317+
o["usage"] = usage_view(in->user_usage_);
1318+
out = std::move(o);
1319+
return true;
1320+
}
1321+
1322+
bool manager::reset_usage(const std::string& id, const std::string& user,
1323+
json::value& out, std::string& err)
1324+
{
1325+
bool changed = false;
1326+
json::object usage;
1327+
{
1328+
std::lock_guard<std::mutex> lock(m_mu_);
1329+
auto in = find_instance_unlocked(id);
1330+
if (!in) {
1331+
err = "not found";
1332+
return false;
1333+
}
1334+
// 仅清零展示口径 rx/tx，配额累计 total 保留（不因统计重置而放开配额）。
1335+
if (user.empty()) {
1336+
for (auto& [_, rec] : in->user_usage_) {
1337+
if (rec.rx || rec.tx) {
1338+
rec.rx = 0;
1339+
rec.tx = 0;
1340+
changed = true;
1341+
}
1342+
}
1343+
} else if (auto it = in->user_usage_.find(user); it != in->user_usage_.end()) {
1344+
if (it->second.rx || it->second.tx) {
1345+
it->second.rx = 0;
1346+
it->second.tx = 0;
1347+
changed = true;
1348+
}
1349+
}
1350+
usage = usage_view(in->user_usage_);
1351+
}
1352+
// 重置需立即落盘，不走节流。
1353+
if (changed)
1354+
save();
1355+
json::object o;
1356+
o["ok"] = true;
1357+
o["usage"] = std::move(usage);
12581358
out = std::move(o);
12591359
return true;
12601360
}
@@ -1313,10 +1413,14 @@ void manager::ws_attached(const instance_ptr& in, jsonrpc_session sess)
13131413
old.stop();
13141414

13151415
// 续接持久化的用户已用量，使配额在重启/重连后延续（proxy 侧只增不减）。
1416+
// 配额口径为合并总量 total（累计上下行由 launcher 单独维护）。
13161417
json::object usage;
13171418
{
13181419
std::lock_guard<std::mutex> lock(m_mu_);
1319-
usage = json::object(in->user_usage_);
1420+
for (const auto& [user, rec] : in->user_usage_) {
1421+
if (rec.total > 0)
1422+
usage[user] = rec.total;
1423+
}
13201424
}
13211425
if (!usage.empty()) {
13221426
jsonrpc_session cur;
@@ -1355,7 +1459,8 @@ void manager::handle_notify(const instance_ptr& in, const std::string& method,
13551459
std::lock_guard<std::mutex> lock(m_mu_);
13561460
in->last_report_ = params;
13571461
in->last_seen_ = now_time();
1358-
// 记录用户已用量（usage_total 为含续接基线的累计总流量），重启后续接。
1462+
// 记录用户已用量：total 为含续接基线的累计总流量（配额续接），
1463+
// rx/tx 为按会话级原始值折算的累计上下行（实例重启归零后仍延续）。
13591464
// 兼容旧版 proxy_server：未上报 usage_total 时回退到会话级 tx+rx。
13601465
bool usage_changed = false;
13611466
if (params.is_object()) {
@@ -1376,16 +1481,26 @@ void manager::handle_notify(const instance_ptr& in, const std::string& method,
13761481
total = as_int64(ut->value());
13771482
if (total <= 0) {
13781483
// 兼容旧版 proxy_server：未上报 usage_total 时回退到会话级 tx+rx。
1379-
auto tx_it = uo.find("tx_bytes");
1380-
auto rx_it = uo.find("rx_bytes");
1381-
total = (tx_it != uo.end() ? as_int64(tx_it->value()) : 0)
1382-
+ (rx_it != uo.end() ? as_int64(rx_it->value()) : 0);
1484+
total = obj_num(uo, "tx_bytes") + obj_num(uo, "rx_bytes");
1485+
}
1486+
// 会话级原始值单调递增；回退说明实例重启，增量从当前值起算。
1487+
std::int64_t raw_rx = obj_num(uo, "rx_bytes");
1488+
std::int64_t raw_tx = obj_num(uo, "tx_bytes");
1489+
auto& rec = in->user_usage_[user];
1490+
if (total > rec.total) {
1491+
rec.total = total;
1492+
usage_changed = true;
1493+
}
1494+
std::int64_t delta_rx = raw_rx >= rec.last_rx ? raw_rx - rec.last_rx : raw_rx;
1495+
std::int64_t delta_tx = raw_tx >= rec.last_tx ? raw_tx - rec.last_tx : raw_tx;
1496+
if (delta_rx > 0 || delta_tx > 0) {
1497+
rec.rx += delta_rx;
1498+
rec.tx += delta_tx;
1499+
usage_changed = true;
13831500
}
1384-
std::int64_t cur = 0;
1385-
if (auto it = in->user_usage_.find(user); it != in->user_usage_.end())
1386-
cur = as_int64(it->value());
1387-
if (total > cur) {
1388-
in->user_usage_[user] = total;
1501+
if (raw_rx != rec.last_rx || raw_tx != rec.last_tx) {
1502+
rec.last_rx = raw_rx;
1503+
rec.last_tx = raw_tx;
13891504
usage_changed = true;
13901505
}
13911506
}

‎apps/launcher/manager.hpp‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,20 @@ namespace net = boost::asio;
3737

3838
using instance_ptr = std::shared_ptr<struct instance>;
3939

40+
// 单个用户的累计流量（字节），随实例持久化。
41+
// total 含续接基线的累计总流量（上行+下行），用于配额续接；重置不清零。
42+
// rx/tx 累计上行（上传）/下行（下载）；重置仅清零这两个展示口径。
43+
// last_rx/last_tx 最近一次会话级原始上报值，用于把重启归零的会话计数
44+
// 折算成增量（回退即视为实例重启）。
45+
struct usage_record
46+
{
47+
std::int64_t total = 0;
48+
std::int64_t rx = 0;
49+
std::int64_t tx = 0;
50+
std::int64_t last_rx = 0;
51+
std::int64_t last_tx = 0;
52+
};
53+
4054
// 一个被管理的 proxy_server 实例。
4155
struct instance : public std::enable_shared_from_this<instance>
4256
{
@@ -47,7 +61,7 @@ struct instance : public std::enable_shared_from_this<instance>
4761
bool autostart_ = false;
4862
time_point created_at_;
4963
// 各用户累计流量用量（字节），持久化；重启实例后续接配额计数。
50-
boost::json::object user_usage_;
64+
std::map<std::string, usage_record> user_usage_;
5165
// 控制通道认证令牌（持久化；launcher 崩溃后孤儿进程可凭旧 token 重连）。
5266
std::string token_;
5367

@@ -180,6 +194,11 @@ class manager : public std::enable_shared_from_this<manager>
180194
boost::json::value summaries();
181195
bool view(const std::string& id, view& out);
182196
bool status_view(const std::string& id, boost::json::value& out);
197+
198+
// 重置用户累计上传/下载（user 为空表示重置该实例全部用户）。返回是否成功；
199+
// out 为重置后的用量视图（{ok, usage}）。配额累计（total）不随重置清零。
200+
bool reset_usage(const std::string& id, const std::string& user,
201+
boost::json::value& out, std::string& err);
183202
// 日志增量。since<0 时返回最近快照（含每行序号）。
184203
bool logs(const std::string& id, std::int64_t since, boost::json::value& out);
185204

0 commit comments

Comments
 (0)