feat(stats): accurate Xray node traffic via outbound stats

Author: ddddeadispose

cc-agent/api.go

@@ -15,7 +15,6 @@ import (
 type API struct {
 	cfg        *Config
 	userStore  *UserStore
-	statsStore *StatsStore
 	xrayClient *XrayClient
 }
 
@@ -192,23 +191,38 @@ func (a *API) handleRemoveUser(w http.ResponseWriter, r *http.Request) {
 	jsonOK(w, map[string]string{"status": "ok"})
 }
 
-// GET /stats — collect fresh stats from Xray, return accumulated totals, then reset
+// GET /stats — take one atomic snapshot from Xray (reset=true) and return
+// { users: { <userId>: {tx, rx} }, node: {tx, rx} }.
+// Users carry per-user uplink/downlink; node is the sum of outbound uplink/downlink
+// across all non-API outbounds (i.e. real traffic that traversed Xray).
 func (a *API) handleStats(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second)
 	defer cancel()
 
-	if err := a.statsStore.CollectFromXray(ctx, a.xrayClient); err != nil {
-		log.Printf("[stats] CollectFromXray: %v", err)
+	rawStats, err := a.xrayClient.QueryStats(ctx, "", true)
+	if err != nil {
+		log.Printf("[stats] QueryStats: %v", err)
+		jsonOK(w, map[string]any{
+			"users": map[string]any{},
+			"node":  map[string]int64{"tx": 0, "rx": 0},
+		})
+		return
 	}
 
-	raw := a.statsStore.GetAndReset()
+	snap := ParseSnapshot(rawStats)
 
-	result := make(map[string]map[string]int64, len(raw))
-	for email, t := range raw {
-		result[email] = map[string]int64{"tx": t.Tx, "rx": t.Rx}
+	users := make(map[string]map[string]int64, len(snap.Users))
+	for email, t := range snap.Users {
+		if t.Tx == 0 && t.Rx == 0 {
+			continue
+		}
+		users[email] = map[string]int64{"tx": t.Tx, "rx": t.Rx}
 	}
 
-	jsonOK(w, result)
+	jsonOK(w, map[string]any{
+		"users": users,
+		"node":  map[string]int64{"tx": snap.Node.Tx, "rx": snap.Node.Rx},
+	})
 }
 
 // POST /restart — restart Xray service, then restore all users (Xray loses state on restart)

cc-agent/main.go

@@ -12,7 +12,7 @@ import (
 	"time"
 )
 
-const Version = "1.0.0"
+const Version = "2.0.0"
 
 var startTime = time.Now()
 
@@ -36,14 +36,15 @@ func main() {
 		log.Printf("[main] Warning: could not load users from disk: %v", err)
 	}
 
-	statsStore := NewStatsStore()
-
 	xrayClient, err := NewXrayClient(cfg)
 	if err != nil {
 		log.Printf("[main] Warning: could not connect to Xray gRPC: %v (will retry on use)", err)
 	}
 
-	// Restore users to Xray after brief startup delay (Xray might still be starting)
+	// Restore users to Xray after a brief startup delay (Xray might still be starting),
+	// then discard the first Xray stats snapshot. Xray accumulates counters from boot,
+	// and without this discard the panel's first /stats poll would attribute all of
+	// that since-boot traffic to the current interval, producing a cold-start spike.
 	go func() {
 		time.Sleep(3 * time.Second)
 		ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
@@ -54,17 +55,10 @@ func main() {
 		} else {
 			log.Printf("[main] Restored %d users to Xray", count)
 		}
-	}()
-
-	// Periodic stats collection from Xray every 60s
-	go func() {
-		for {
-			time.Sleep(60 * time.Second)
-			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-			if err := statsStore.CollectFromXray(ctx, xrayClient); err != nil {
-				log.Printf("[stats] Collection error: %v", err)
-			}
-			cancel()
+		if _, err := xrayClient.QueryStats(ctx, "", true); err != nil {
+			log.Printf("[main] Startup stats discard failed: %v", err)
+		} else {
+			log.Printf("[main] Startup stats discarded (cold-start reset)")
 		}
 	}()
 
@@ -81,7 +75,6 @@ func main() {
 	api := &API{
 		cfg:        cfg,
 		userStore:  userStore,
-		statsStore: statsStore,
 		xrayClient: xrayClient,
 	}
 

cc-agent/stats.go

@@ -1,74 +1,70 @@
 package main
 
-import (
-	"context"
-	"log"
-	"strings"
-	"sync"
-)
+import "strings"
 
-// UserTraffic holds accumulated uplink/downlink bytes for a user
+// UserTraffic holds uplink/downlink bytes.
+// For users: Tx = client uplink (client -> server), Rx = client downlink (server -> client).
+// For the node: Tx/Rx are the sum of outbound uplink/downlink across all non-API outbounds,
+// which is the real traffic that actually traversed Xray.
 type UserTraffic struct {
 	Tx int64 `json:"tx"` // uplink bytes
 	Rx int64 `json:"rx"` // downlink bytes
 }
 
-// StatsStore accumulates traffic stats between panel polls
-type StatsStore struct {
-	mu      sync.Mutex
-	traffic map[string]*UserTraffic // keyed by email
+// Snapshot is the parsed result of a single Xray QueryStats call with an empty
+// pattern. Users is keyed by email (which equals the panel userId). Node is the
+// aggregated node-level traffic derived from outbound stats.
+type Snapshot struct {
+	Users map[string]*UserTraffic
+	Node  UserTraffic
 }
 
-func NewStatsStore() *StatsStore {
-	return &StatsStore{
-		traffic: make(map[string]*UserTraffic),
-	}
-}
+// apiOutboundTag is the tag of the internal Xray API outbound; its counters
+// represent gRPC control-plane traffic and must not be attributed to the node.
+const apiOutboundTag = "API"
 
-// CollectFromXray fetches stats from Xray (with reset) and adds to local accumulator.
-// Xray resets its own counters after each query with reset=true.
-func (s *StatsStore) CollectFromXray(ctx context.Context, xc *XrayClient) error {
-	rawStats, err := xc.QueryStats(ctx, true)
-	if err != nil {
-		return err
+// ParseSnapshot converts a flat map of Xray stat names to a structured Snapshot.
+// Expected stat name formats:
+//   - user>>>{email}>>>traffic>>>{uplink|downlink}
+//   - outbound>>>{tag}>>>traffic>>>{uplink|downlink}
+//   - inbound>>>{tag}>>>traffic>>>{uplink|downlink}   (ignored here; reserved for future metrics)
+func ParseSnapshot(rawStats map[string]int64) Snapshot {
+	snap := Snapshot{
+		Users: make(map[string]*UserTraffic, len(rawStats)/2),
 	}
 
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	for name, value := range rawStats {
-		// Name format: user>>>email>>>traffic>>>uplink or downlink
 		parts := strings.Split(name, ">>>")
-		if len(parts) != 4 || parts[0] != "user" || parts[2] != "traffic" {
+		if len(parts) != 4 || parts[2] != "traffic" {
 			continue
 		}
-		email := parts[1]
-		direction := parts[3]
+		kind, id, direction := parts[0], parts[1], parts[3]
 
-		if s.traffic[email] == nil {
-			s.traffic[email] = &UserTraffic{}
-		}
-		switch direction {
-		case "uplink":
-			s.traffic[email].Tx += value
-		case "downlink":
-			s.traffic[email].Rx += value
+		switch kind {
+		case "user":
+			ut := snap.Users[id]
+			if ut == nil {
+				ut = &UserTraffic{}
+				snap.Users[id] = ut
+			}
+			switch direction {
+			case "uplink":
+				ut.Tx += value
+			case "downlink":
+				ut.Rx += value
+			}
+		case "outbound":
+			if id == apiOutboundTag {
+				continue
+			}
+			switch direction {
+			case "uplink":
+				snap.Node.Tx += value
+			case "downlink":
+				snap.Node.Rx += value
+			}
 		}
 	}
 
-	if len(rawStats) > 0 {
-		log.Printf("[stats] Collected %d stat entries from Xray", len(rawStats))
-	}
-	return nil
-}
-
-// GetAndReset returns all accumulated stats and resets the local store.
-// Called when the panel polls /stats.
-func (s *StatsStore) GetAndReset() map[string]*UserTraffic {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	result := s.traffic
-	s.traffic = make(map[string]*UserTraffic)
-	return result
+	return snap
 }

cc-agent/xray.go

@@ -72,10 +72,12 @@ func (c *XrayClient) RemoveUser(ctx context.Context, email string) error {
 	return nil
 }
 
-// QueryStats fetches traffic stats from Xray; reset=true clears counters in Xray
-func (c *XrayClient) QueryStats(ctx context.Context, reset bool) (map[string]int64, error) {
+// QueryStats fetches traffic stats from Xray matching the given pattern.
+// An empty pattern returns all stats (users, inbounds, outbounds) in one call.
+// reset=true atomically clears the matched counters in Xray.
+func (c *XrayClient) QueryStats(ctx context.Context, pattern string, reset bool) (map[string]int64, error) {
 	resp, err := c.stats.QueryStats(ctx, &stats_command.QueryStatsRequest{
-		Pattern: "user>>>",
+		Pattern: pattern,
 		Reset_:  reset,
 	})
 	if err != nil {

src/services/configGenerator.js

@@ -507,6 +507,8 @@ function generateXrayConfig(node, users) {
             system: {
                 statsInboundUplink: true,
                 statsInboundDownlink: true,
+                statsOutboundUplink: true,
+                statsOutboundDownlink: true,
             },
         },
         inbounds: [

src/services/syncService.js

@@ -377,7 +377,10 @@ class SyncService {
 
     /**
      * Collect traffic stats from Xray node via Agent GET /stats.
-     * Agent accumulates stats between polls (Xray counters are reset on each agent collection).
+     * Response shape (agent v2.0.0+):
+     *   { users: { <userId>: { tx, rx } }, node: { tx, rx } }
+     * Node tx/rx come from Xray outbound stats (real traffic that traversed
+     * Xray), not from summing per-user counters.
      */
     async collectXrayTrafficStats(node) {
         if (!(node.xray?.agentToken)) {
@@ -387,23 +390,23 @@ class SyncService {
 
         try {
             const response = await this._agentRequest(node, 'GET', '/stats');
-            const stats = response.data || {};
+            const data = response.data || {};
+            const users = data.users || {};
+            const nodeTraffic = data.node || { tx: 0, rx: 0 };
+            const nodeTx = nodeTraffic.tx || 0;
+            const nodeRx = nodeTraffic.rx || 0;
 
-            if (Object.keys(stats).length === 0) return;
+            const userEntries = Object.entries(users);
+            if (userEntries.length === 0 && nodeTx === 0 && nodeRx === 0) return;
 
-            let nodeTx = 0;
-            let nodeRx = 0;
             const bulkOps = [];
             const now = new Date();
 
-            for (const [email, traffic] of Object.entries(stats)) {
+            for (const [email, traffic] of userEntries) {
                 const tx = traffic.tx || 0;
                 const rx = traffic.rx || 0;
                 if (tx === 0 && rx === 0) continue;
 
-                nodeTx += tx;
-                nodeRx += rx;
-
                 // email == userId (as set in configGenerator and agent)
                 bulkOps.push({
                     updateOne: {
@@ -419,7 +422,7 @@ class SyncService {
             if (bulkOps.length > 0) {
                 const result = await HyUser.bulkWrite(bulkOps, { ordered: false });
                 logger.debug(`[Agent Stats] ${node.name}: updated ${result.modifiedCount}/${bulkOps.length} users`);
-                this._checkUserLimits(Object.keys(stats)).catch(() => {});
+                this._checkUserLimits(userEntries.map(([email]) => email)).catch(() => {});
             }
 
             if (nodeTx > 0 || nodeRx > 0) {
@@ -430,7 +433,7 @@ class SyncService {
                         $set: { 'traffic.lastUpdate': now },
                     }
                 );
-                logger.info(`[Agent Stats] ${node.name}: ${bulkOps.length} users, ↑${(nodeTx / 1024 / 1024).toFixed(1)}MB ↓${(nodeRx / 1024 / 1024).toFixed(1)}MB`);
+                logger.info(`[Agent Stats] ${node.name}: ${bulkOps.length} users, node ↑${(nodeTx / 1024 / 1024).toFixed(1)}MB ↓${(nodeRx / 1024 / 1024).toFixed(1)}MB`);
             }
         } catch (error) {
             logger.error(`[Agent Stats] ${node.name} error: ${error.message}`);