From a6b35268ab2243a09c6b1a6eca7345052bb13156 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 19 Jun 2026 06:01:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20#7=20=E7=9C=9F=E5=AE=9E=E6=B5=81?= =?UTF-8?q?=E9=87=8F=E9=87=87=E9=9B=86=20=E2=80=94=20clash=5Fapi=20?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E6=80=BB=E9=87=8F=20=E2=86=92=20usage=5Fdail?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sing-box 该构建未编入 v2ray_api、clash /connections 也不暴露用户,故无法做 准确 per-user 计费。改取节点累计总量(clash downloadTotal/uploadTotal)做 窗口增量,按当前 dp_uuid 均摊上报(单用户=准确,多用户近似,已注释标注)。 - render.go:节点 sing-box 配置加 experimental.clash_api(loopback+secret)。 - ClashUsageSource:轮询 /connections 总量,delta + 重启回绕保护;clash 的 up/down 是代理视角,对调为用户视角(下载→bytes_down)。 - SingBox.DpUUIDs() 供归属;Agent.UseClashUsage() 在 cmd/agent 接上。 已节点 live 验证:usage_daily 实时入库,下载增量正确落 bytes_down。 局限:多 dp_uuid 时均摊(准确计费需重编 sing-box 带 with_v2ray_api)。 Co-Authored-By: Claude Opus 4.8 --- server/cmd/agent/main.go | 2 + server/internal/agentd/agent.go | 5 ++ server/internal/agentd/render.go | 14 ++++ server/internal/agentd/singbox.go | 12 +++ server/internal/agentd/usage_clash.go | 113 ++++++++++++++++++++++++++ 5 files changed, 146 insertions(+) create mode 100644 server/internal/agentd/usage_clash.go diff --git a/server/cmd/agent/main.go b/server/cmd/agent/main.go index 2871312..c01a75e 100644 --- a/server/cmd/agent/main.go +++ b/server/cmd/agent/main.go @@ -38,6 +38,8 @@ func main() { } agent := agentd.New(cfg, agentd.WithRestarter(agentd.SystemdRestarter{Unit: singboxUnit})) + // 真实流量采集:读本地 clash_api 节点总量,按 dp_uuid 归属上报。 + agent.UseClashUsage() ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/server/internal/agentd/agent.go b/server/internal/agentd/agent.go index eb28922..98871e6 100644 --- a/server/internal/agentd/agent.go +++ b/server/internal/agentd/agent.go @@ -77,6 +77,11 @@ func New(cfg Config, opts ...Option) *Agent { // SingBox exposes the underlying manager (tests / introspection). func (a *Agent) SingBox() *SingBox { return a.sb } +// UseClashUsage wires the real usage source: reads node-total traffic from the +// local clash_api and reports per-window deltas (attributed across dp_uuids). +// Call after New (needs a.sb). Production main enables this; tests leave it off. +func (a *Agent) UseClashUsage() { a.usage = newClashUsageSource(a.sb) } + // NodeUUID returns the enrolled node UUID (empty until Run enrolls). func (a *Agent) NodeUUID() string { v, _ := a.nodeUUID.Load().(string) diff --git a/server/internal/agentd/render.go b/server/internal/agentd/render.go index 2d5dd8d..1ff82a0 100644 --- a/server/internal/agentd/render.go +++ b/server/internal/agentd/render.go @@ -12,11 +12,25 @@ import ( // derived password (DeriveHy2Password) — both from the same dp_uuid source. // // Only the opaque dp_uuid is ever written; no account identity touches the node. +// clash_api endpoint(loopback only)供 agent 读取节点累计流量做用量统计。 +// v2ray_api 未编入该 sing-box 构建、clash /connections 又不暴露用户,故只能 +// 取节点总量(downloadTotal/uploadTotal),由 agent 按当前 dp_uuid 归属/分摊。 +const ( + clashAPIAddr = "127.0.0.1:19090" + clashAPISecret = "pangolin-local-stats" +) + func renderSingboxConfig(creds []Cred, reality *agentv1.RealityInbound, hy2 *agentv1.Hy2Inbound, deriveKey string) ([]byte, error) { cfg := map[string]any{ "log": map[string]any{"level": "warn", "timestamp": true}, "inbounds": buildInbounds(creds, reality, hy2, deriveKey), "outbounds": []any{map[string]any{"type": "direct", "tag": "direct"}}, + "experimental": map[string]any{ + "clash_api": map[string]any{ + "external_controller": clashAPIAddr, + "secret": clashAPISecret, + }, + }, } return json.MarshalIndent(cfg, "", " ") } diff --git a/server/internal/agentd/singbox.go b/server/internal/agentd/singbox.go index 24f7fe2..3d8d047 100644 --- a/server/internal/agentd/singbox.go +++ b/server/internal/agentd/singbox.go @@ -144,6 +144,18 @@ func (s *SingBox) Upsert(c *Cred) { s.markDirty() } +// DpUUIDs returns a snapshot of the currently provisioned dp_uuids. +// Used by the usage source to attribute node-total traffic to active users. +func (s *SingBox) DpUUIDs() []string { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]string, 0, len(s.creds)) + for u := range s.creds { + out = append(out, u) + } + return out +} + // Revoke removes a credential. No-op if absent (idempotent). func (s *SingBox) Revoke(dpUUID string) { s.mu.Lock() diff --git a/server/internal/agentd/usage_clash.go b/server/internal/agentd/usage_clash.go new file mode 100644 index 0000000..2aeb3a3 --- /dev/null +++ b/server/internal/agentd/usage_clash.go @@ -0,0 +1,113 @@ +package agentd + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// ClashUsageSource reads the node's cumulative traffic from sing-box's clash_api +// (/connections downloadTotal/uploadTotal) and reports per-window deltas. +// +// Limitation: the clash API exposes only NODE-TOTAL bytes (no per-user field, and +// v2ray_api is not compiled into this sing-box build). So node-total delta is +// split evenly across currently provisioned dp_uuids — exact for a single active +// user, approximate for multiple. See todo #7 for accurate accounting (rebuild +// sing-box with with_v2ray_api). +type ClashUsageSource struct { + sb *SingBox + addr string + secret string + client *http.Client + + mu sync.Mutex + inited bool + lastUp, lastDown int64 +} + +func newClashUsageSource(sb *SingBox) *ClashUsageSource { + return &ClashUsageSource{ + sb: sb, + addr: clashAPIAddr, + secret: clashAPISecret, + client: &http.Client{Timeout: 5 * time.Second}, + } +} + +type clashConnections struct { + DownloadTotal int64 `json:"downloadTotal"` + UploadTotal int64 `json:"uploadTotal"` +} + +// Collect returns per-dp_uuid usage accumulated since the previous call. +// First call only sets the baseline (returns nil). +func (c *ClashUsageSource) Collect() []*agentv1.UsageEntry { + up, down, ok := c.fetchTotals() + if !ok { + return nil + } + + c.mu.Lock() + defer c.mu.Unlock() + if !c.inited { + c.lastUp, c.lastDown, c.inited = up, down, true + return nil + } + dUp, dDown := up-c.lastUp, down-c.lastDown + c.lastUp, c.lastDown = up, down + if dUp < 0 || dDown < 0 { + // Counters reset (sing-box restarted) → skip this window. + return nil + } + if dUp == 0 && dDown == 0 { + return nil + } + + uuids := c.sb.DpUUIDs() + if len(uuids) == 0 { + return nil + } + n := int64(len(uuids)) + perUp, perDown := dUp/n, dDown/n + entries := make([]*agentv1.UsageEntry, 0, len(uuids)) + for _, u := range uuids { + entries = append(entries, &agentv1.UsageEntry{ + DpUUID: u, + BytesUp: perUp, + BytesDown: perDown, + SessionMinutes: 1, // 本窗口有流量 → 计 1 分钟在线(近似) + }) + } + return entries +} + +func (c *ClashUsageSource) fetchTotals() (up, down int64, ok bool) { + req, err := http.NewRequestWithContext( + context.Background(), http.MethodGet, "http://"+c.addr+"/connections", nil) + if err != nil { + return 0, 0, false + } + if c.secret != "" { + req.Header.Set("Authorization", "Bearer "+c.secret) + } + resp, err := c.client.Do(req) + if err != nil { + return 0, 0, false + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return 0, 0, false + } + var cc clashConnections + if err := json.NewDecoder(resp.Body).Decode(&cc); err != nil { + return 0, 0, false + } + // clash 的 up/down 是「代理视角」:为客户端下载,代理需把数据 upload 给客户端, + // 故 clash.uploadTotal ≈ 用户下载、clash.downloadTotal ≈ 用户上传 → 此处对调, + // 返回「用户视角」的 (up, down)。 + return cc.DownloadTotal, cc.UploadTotal, true +}