Add v2ray user stats api

adapter/experimental.go

@@ -45,6 +45,6 @@ type V2RayServer interface {
 }
 
 type V2RayStatsService interface {
-	RoutedConnection(inbound string, outbound string, conn net.Conn) net.Conn
+	RoutedConnection(inbound string, outbound string, user string, conn net.Conn) net.Conn
-	RoutedPacketConnection(inbound string, outbound string, conn N.PacketConn) N.PacketConn
+	RoutedPacketConnection(inbound string, outbound string, user string, conn N.PacketConn) N.PacketConn
 }

docs/configuration/experimental/index.md

@@ -25,6 +25,9 @@
         "outbounds": [
           "proxy",
           "direct"
+        ],
+        "users": [
+          "sekai"
         ]
       }
     }
@@ -109,3 +112,7 @@ Inbound list to count traffic.
 #### stats.outbounds
 
 Outbound list to count traffic.
+
+#### stats.users
+
+User list to count traffic.

docs/configuration/experimental/index.zh.md

@@ -25,6 +25,9 @@
         "outbounds": [
           "proxy",
           "direct"
+        ],
+        "users": [
+          "sekai"
         ]
       }
     }
@@ -107,3 +110,7 @@ gRPC API 监听地址。如果为空，则禁用 V2Ray API。
 #### stats.outbounds
 
 统计流量的出站列表。
+
+#### stats.users
+
+统计流量的用户列表。

experimental/v2rayapi/stats.go

@@ -31,6 +31,7 @@ type StatsService struct {
 	createdAt time.Time
 	inbounds  map[string]bool
 	outbounds map[string]bool
+	users     map[string]bool
 	access    sync.Mutex
 	counters  map[string]*atomic.Int64
 }
@@ -41,26 +42,32 @@ func NewStatsService(options option.V2RayStatsServiceOptions) *StatsService {
 	}
 	inbounds := make(map[string]bool)
 	outbounds := make(map[string]bool)
+	users := make(map[string]bool)
 	for _, inbound := range options.Inbounds {
 		inbounds[inbound] = true
 	}
 	for _, outbound := range options.Outbounds {
 		outbounds[outbound] = true
 	}
+	for _, user := range options.Users {
+		users[user] = true
+	}
 	return &StatsService{
 		createdAt: time.Now(),
 		inbounds:  inbounds,
 		outbounds: outbounds,
+		users:     users,
 		counters:  make(map[string]*atomic.Int64),
 	}
 }
 
-func (s *StatsService) RoutedConnection(inbound string, outbound string, conn net.Conn) net.Conn {
+func (s *StatsService) RoutedConnection(inbound string, outbound string, user string, conn net.Conn) net.Conn {
 	var readCounter []*atomic.Int64
 	var writeCounter []*atomic.Int64
 	countInbound := inbound != "" && s.inbounds[inbound]
 	countOutbound := outbound != "" && s.outbounds[outbound]
-	if !countInbound && !countOutbound {
+	countUser := user != "" && s.users[user]
+	if !countInbound && !countOutbound && !countUser {
 		return conn
 	}
 	s.access.Lock()
@@ -72,16 +79,21 @@ func (s *StatsService) RoutedConnection(inbound string, outbound string, conn ne
 		readCounter = append(readCounter, s.loadOrCreateCounter("outbound>>>"+outbound+">>>traffic>>>uplink"))
 		writeCounter = append(writeCounter, s.loadOrCreateCounter("outbound>>>"+outbound+">>>traffic>>>downlink"))
 	}
+	if countUser {
+		readCounter = append(readCounter, s.loadOrCreateCounter("user>>>"+user+">>>traffic>>>uplink"))
+		writeCounter = append(writeCounter, s.loadOrCreateCounter("user>>>"+user+">>>traffic>>>downlink"))
+	}
 	s.access.Unlock()
 	return trackerconn.New(conn, readCounter, writeCounter)
 }
 
-func (s *StatsService) RoutedPacketConnection(inbound string, outbound string, conn N.PacketConn) N.PacketConn {
+func (s *StatsService) RoutedPacketConnection(inbound string, outbound string, user string, conn N.PacketConn) N.PacketConn {
 	var readCounter []*atomic.Int64
 	var writeCounter []*atomic.Int64
 	countInbound := inbound != "" && s.inbounds[inbound]
 	countOutbound := outbound != "" && s.outbounds[outbound]
-	if !countInbound && !countOutbound {
+	countUser := user != "" && s.users[user]
+	if !countInbound && !countOutbound && !countUser {
 		return conn
 	}
 	s.access.Lock()
@@ -93,6 +105,10 @@ func (s *StatsService) RoutedPacketConnection(inbound string, outbound string, c
 		readCounter = append(readCounter, s.loadOrCreateCounter("outbound>>>"+outbound+">>>traffic>>>uplink"))
 		writeCounter = append(writeCounter, s.loadOrCreateCounter("outbound>>>"+outbound+">>>traffic>>>downlink"))
 	}
+	if countUser {
+		readCounter = append(readCounter, s.loadOrCreateCounter("user>>>"+user+">>>traffic>>>uplink"))
+		writeCounter = append(writeCounter, s.loadOrCreateCounter("user>>>"+user+">>>traffic>>>downlink"))
+	}
 	s.access.Unlock()
 	return trackerconn.NewPacket(conn, readCounter, writeCounter)
 }

option/v2ray.go

@@ -9,4 +9,5 @@ type V2RayStatsServiceOptions struct {
 	Enabled   bool     `json:"enabled,omitempty"`
 	Inbounds  []string `json:"inbounds,omitempty"`
 	Outbounds []string `json:"outbounds,omitempty"`
+	Users     []string `json:"users,omitempty"`
 }

route/router.go

@@ -600,7 +600,7 @@ func (r *Router) RouteConnection(ctx context.Context, conn net.Conn, metadata ad
 	}
 	if r.v2rayServer != nil {
 		if statsService := r.v2rayServer.StatsService(); statsService != nil {
-			conn = statsService.RoutedConnection(metadata.Inbound, detour.Tag(), conn)
+			conn = statsService.RoutedConnection(metadata.Inbound, detour.Tag(), metadata.User, conn)
 		}
 	}
 	return detour.NewConnection(ctx, conn, metadata)
@@ -678,7 +678,7 @@ func (r *Router) RoutePacketConnection(ctx context.Context, conn N.PacketConn, m
 	}
 	if r.v2rayServer != nil {
 		if statsService := r.v2rayServer.StatsService(); statsService != nil {
-			conn = statsService.RoutedPacketConnection(metadata.Inbound, detour.Tag(), conn)
+			conn = statsService.RoutedPacketConnection(metadata.Inbound, detour.Tag(), metadata.User, conn)
 		}
 	}
 	return detour.NewPacketConnection(ctx, conn, metadata)