feat(agent): #7 真实流量采集 — clash_api 节点总量 → usage_daily

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 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-19 06:01:24 +08:00
parent f35bbea0a0
commit a6b35268ab
5 changed files with 146 additions and 0 deletions
+2
View File
@@ -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()
+5
View File
@@ -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)
+14
View File
@@ -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, "", " ")
}
+12
View File
@@ -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()
+113
View File
@@ -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
}