test: 增加MinReportTraffic最低流量上报阈值

common/counter/traffic.go

@@ -6,7 +6,7 @@ import (
 )
 
 type TrafficCounter struct {
-	counters sync.Map
+	Counters sync.Map
 }
 
 type TrafficStorage struct {
@@ -18,26 +18,26 @@ func NewTrafficCounter() *TrafficCounter {
 	return &TrafficCounter{}
 }
 
-func (c *TrafficCounter) GetCounter(id string) *TrafficStorage {
+func (c *TrafficCounter) GetCounter(uuid string) *TrafficStorage {
-	if cts, ok := c.counters.Load(id); ok {
+	if cts, ok := c.Counters.Load(uuid); ok {
 		return cts.(*TrafficStorage)
 	}
 	newStorage := &TrafficStorage{}
-	if cts, loaded := c.counters.LoadOrStore(id, newStorage); loaded {
+	if cts, loaded := c.Counters.LoadOrStore(uuid, newStorage); loaded {
 		return cts.(*TrafficStorage)
 	}
 	return newStorage
 }
 
-func (c *TrafficCounter) GetUpCount(id string) int64 {
+func (c *TrafficCounter) GetUpCount(uuid string) int64 {
-	if cts, ok := c.counters.Load(id); ok {
+	if cts, ok := c.Counters.Load(uuid); ok {
 		return cts.(*TrafficStorage).UpCounter.Load()
 	}
 	return 0
 }
 
-func (c *TrafficCounter) GetDownCount(id string) int64 {
+func (c *TrafficCounter) GetDownCount(uuid string) int64 {
-	if cts, ok := c.counters.Load(id); ok {
+	if cts, ok := c.Counters.Load(uuid); ok {
 		return cts.(*TrafficStorage).DownCounter.Load()
 	}
 	return 0
@@ -45,30 +45,30 @@ func (c *TrafficCounter) GetDownCount(id string) int64 {
 
 func (c *TrafficCounter) Len() int {
 	length := 0
-	c.counters.Range(func(_, _ interface{}) bool {
+	c.Counters.Range(func(_, _ interface{}) bool {
 		length++
 		return true
 	})
 	return length
 }
 
-func (c *TrafficCounter) Reset(id string) {
+func (c *TrafficCounter) Reset(uuid string) {
-	if cts, ok := c.counters.Load(id); ok {
+	if cts, ok := c.Counters.Load(uuid); ok {
 		cts.(*TrafficStorage).UpCounter.Store(0)
 		cts.(*TrafficStorage).DownCounter.Store(0)
 	}
 }
 
-func (c *TrafficCounter) Delete(id string) {
+func (c *TrafficCounter) Delete(uuid string) {
-	c.counters.Delete(id)
+	c.Counters.Delete(uuid)
 }
 
-func (c *TrafficCounter) Rx(id string, n int) {
+func (c *TrafficCounter) Rx(uuid string, n int) {
-	cts := c.GetCounter(id)
+	cts := c.GetCounter(uuid)
 	cts.DownCounter.Add(int64(n))
 }
 
-func (c *TrafficCounter) Tx(id string, n int) {
+func (c *TrafficCounter) Tx(uuid string, n int) {
-	cts := c.GetCounter(id)
+	cts := c.GetCounter(uuid)
 	cts.UpCounter.Add(int64(n))
 }

conf/node.go

@@ -109,6 +109,7 @@ type Options struct {
 	ListenIP               string          `json:"ListenIP"`
 	SendIP                 string          `json:"SendIP"`
 	DeviceOnlineMinTraffic int64           `json:"DeviceOnlineMinTraffic"`
+	ReportMinTraffic       int64           `json:"ReportMinTraffic"`
 	LimitConfig            LimitConfig     `json:"LimitConfig"`
 	RawOptions             json.RawMessage `json:"RawOptions"`
 	XrayOptions            *XrayOptions    `json:"XrayOptions"`

core/hy2/hook.go

@@ -14,9 +14,10 @@ import (
 var _ server.TrafficLogger = (*HookServer)(nil)
 
 type HookServer struct {
-	Tag     string
+	Tag                   string
-	logger  *zap.Logger
+	logger                *zap.Logger
-	Counter sync.Map
+	Counter               sync.Map
+	ReportMinTrafficBytes int64
 }
 
 func (h *HookServer) TraceStream(stream quic.Stream, stats *server.StreamStats) {

core/hy2/node.go

@@ -40,8 +40,9 @@ func (h *Hysteria2) AddNode(tag string, info *panel.NodeInfo, config *conf.Optio
 			logger: h.Logger,
 		},
 		TrafficLogger: &HookServer{
-			Tag:    tag,
+			Tag:                   tag,
-			logger: h.Logger,
+			logger:                h.Logger,
+			ReportMinTrafficBytes: config.ReportMinTraffic * 1024,
 		},
 	}
 

core/hy2/user.go

@@ -14,12 +14,12 @@ var _ server.Authenticator = &V2bX{}
 
 type V2bX struct {
 	usersMap map[string]int
-	mutex    sync.Mutex
+	mutex    sync.RWMutex
 }
 
 func (v *V2bX) Authenticate(addr net.Addr, auth string, tx uint64) (ok bool, id string) {
-	v.mutex.Lock()
+	v.mutex.RLock()
-	defer v.mutex.Unlock()
+	defer v.mutex.RUnlock()
 	if _, exists := v.usersMap[auth]; exists {
 		return true, auth
 	}
@@ -56,15 +56,38 @@ func (h *Hysteria2) DelUsers(users []panel.UserInfo, tag string, _ *panel.NodeIn
 	return nil
 }
 
-func (h *Hysteria2) GetUserTraffic(tag string, uuid string, reset bool) (up int64, down int64) {
+func (h *Hysteria2) GetUserTrafficSlice(tag string, reset bool) ([]panel.UserTraffic, error) {
-	if v, ok := h.Hy2nodes[tag].TrafficLogger.(*HookServer).Counter.Load(tag); ok {
+	trafficSlice := make([]panel.UserTraffic, 0)
-		c := v.(*counter.TrafficCounter)
+	h.Auth.mutex.RLock()
-		up = c.GetUpCount(uuid)
+	defer h.Auth.mutex.RUnlock()
-		down = c.GetDownCount(uuid)
+	if _, ok := h.Hy2nodes[tag]; !ok {
-		if reset {
+		return nil, nil
-			c.Reset(uuid)
-		}
-		return up, down
 	}
-	return 0, 0
+	hook := h.Hy2nodes[tag].TrafficLogger.(*HookServer)
+	if v, ok := hook.Counter.Load(tag); ok {
+		c := v.(*counter.TrafficCounter)
+		c.Counters.Range(func(key, value interface{}) bool {
+			uuid := key.(string)
+			traffic := value.(*counter.TrafficStorage)
+			up := traffic.UpCounter.Load()
+			down := traffic.DownCounter.Load()
+			if up+down >= hook.ReportMinTrafficBytes {
+				if reset {
+					traffic.UpCounter.Store(0)
+					traffic.DownCounter.Store(0)
+				}
+				trafficSlice = append(trafficSlice, panel.UserTraffic{
+					UID:      h.Auth.usersMap[uuid],
+					Upload:   up,
+					Download: down,
+				})
+			}
+			return true
+		})
+		if len(trafficSlice) == 0 {
+			return nil, nil
+		}
+		return trafficSlice, nil
+	}
+	return nil, nil
 }

core/interface.go

@@ -17,7 +17,7 @@ type Core interface {
 	AddNode(tag string, info *panel.NodeInfo, config *conf.Options) error
 	DelNode(tag string) error
 	AddUsers(p *AddUsersParams) (added int, err error)
-	GetUserTraffic(tag, uuid string, reset bool) (up int64, down int64)
+	GetUserTrafficSlice(tag string, reset bool) ([]panel.UserTraffic, error)
 	DelUsers(users []panel.UserInfo, tag string, info *panel.NodeInfo) error
 	Protocols() []string
 	Type() string

core/selector.go

@@ -128,12 +128,12 @@ func (s *Selector) AddUsers(p *AddUsersParams) (added int, err error) {
 	return t.(Core).AddUsers(p)
 }
 
-func (s *Selector) GetUserTraffic(tag, uuid string, reset bool) (up int64, down int64) {
+func (s *Selector) GetUserTrafficSlice(tag string, reset bool) ([]panel.UserTraffic, error) {
 	t, e := s.nodes.Load(tag)
 	if !e {
-		return 0, 0
+		return nil, errors.New("the node is not have")
 	}
-	return t.(Core).GetUserTraffic(tag, uuid, reset)
+	return t.(Core).GetUserTrafficSlice(tag, reset)
 }
 
 func (s *Selector) DelUsers(users []panel.UserInfo, tag string, info *panel.NodeInfo) error {

core/sing/hook.go

@@ -27,13 +27,6 @@ func (h *HookServer) ModeList() []string {
 	return nil
 }
 
-func NewHookServer() *HookServer {
-	server := &HookServer{
-		counter: sync.Map{},
-	}
-	return server
-}
-
 func (h *HookServer) RoutedConnection(_ context.Context, conn net.Conn, m adapter.InboundContext, _ adapter.Rule, _ adapter.Outbound) net.Conn {
 	l, err := limiter.GetLimiter(m.Inbound)
 	if err != nil {

core/sing/node.go

@@ -394,6 +394,7 @@ func getInboundOptions(tag string, info *panel.NodeInfo, c *conf.Options) (optio
 }
 
 func (b *Sing) AddNode(tag string, info *panel.NodeInfo, config *conf.Options) error {
+	b.nodeReportMinTrafficBytes[tag] = config.ReportMinTraffic * 1024
 	c, err := getInboundOptions(tag, info, config)
 	if err != nil {
 		return err

core/sing/sing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 
 	"github.com/sagernet/sing-box/include"
 	"github.com/sagernet/sing-box/log"
@@ -24,11 +25,18 @@ type DNSConfig struct {
 }
 
 type Sing struct {
-	box        *box.Box
+	box                       *box.Box
-	ctx        context.Context
+	ctx                       context.Context
-	hookServer *HookServer
+	hookServer                *HookServer
-	router     adapter.Router
+	router                    adapter.Router
-	logFactory log.Factory
+	logFactory                log.Factory
+	users                     *UserMap
+	nodeReportMinTrafficBytes map[string]int64
+}
+
+type UserMap struct {
+	uidMap  map[string]int
+	mapLock sync.RWMutex
 }
 
 func init() {
@@ -71,7 +79,9 @@ func New(c *conf.CoreConfig) (vCore.Core, error) {
 	if err != nil {
 		return nil, err
 	}
-	hs := NewHookServer()
+	hs := &HookServer{
+		counter: sync.Map{},
+	}
 	b.Router().AppendTracker(hs)
 	return &Sing{
 		ctx:        b.Router().GetCtx(),
@@ -79,6 +89,10 @@ func New(c *conf.CoreConfig) (vCore.Core, error) {
 		hookServer: hs,
 		router:     b.Router(),
 		logFactory: b.LogFactory(),
+		users: &UserMap{
+			uidMap: make(map[string]int),
+		},
+		nodeReportMinTrafficBytes: make(map[string]int64),
 	}, nil
 }
 

core/sing/user.go

@@ -23,6 +23,11 @@ func (b *Sing) AddUsers(p *core.AddUsersParams) (added int, err error) {
 	if !found {
 		return 0, errors.New("the inbound not found")
 	}
+	b.users.mapLock.Lock()
+	defer b.users.mapLock.Unlock()
+	for i := range p.Users {
+		b.users.uidMap[p.Users[i].Uuid] = p.Users[i].Id
+	}
 	switch p.NodeInfo.Type {
 	case "vless":
 		us := make([]option.VLESSUser, len(p.Users))
@@ -129,6 +134,39 @@ func (b *Sing) GetUserTraffic(tag, uuid string, reset bool) (up int64, down int6
 	return 0, 0
 }
 
+func (b *Sing) GetUserTrafficSlice(tag string, reset bool) ([]panel.UserTraffic, error) {
+	trafficSlice := make([]panel.UserTraffic, 0)
+	hook := b.hookServer
+	b.users.mapLock.RLock()
+	defer b.users.mapLock.RUnlock()
+	if v, ok := hook.counter.Load(tag); ok {
+		c := v.(*counter.TrafficCounter)
+		c.Counters.Range(func(key, value interface{}) bool {
+			uuid := key.(string)
+			traffic := value.(*counter.TrafficStorage)
+			up := traffic.UpCounter.Load()
+			down := traffic.DownCounter.Load()
+			if up+down >= b.nodeReportMinTrafficBytes[tag] {
+				if reset {
+					traffic.UpCounter.Store(0)
+					traffic.DownCounter.Store(0)
+				}
+				trafficSlice = append(trafficSlice, panel.UserTraffic{
+					UID:      b.users.uidMap[uuid],
+					Upload:   up,
+					Download: down,
+				})
+			}
+			return true
+		})
+		if len(trafficSlice) == 0 {
+			return nil, nil
+		}
+		return trafficSlice, nil
+	}
+	return nil, nil
+}
+
 type UserDeleter interface {
 	DelUsers(uuid []string) error
 }
@@ -158,7 +196,10 @@ func (b *Sing) DelUsers(users []panel.UserInfo, tag string, info *panel.NodeInfo
 		return errors.New("the inbound not found")
 	}
 	uuids := make([]string, len(users))
+	b.users.mapLock.Lock()
+	defer b.users.mapLock.Unlock()
 	for i := range users {
+		delete(b.users.uidMap, users[i].Uuid)
 		uuids[i] = users[i].Uuid
 	}
 	err := del.DelUsers(uuids)

core/xray/app/dispatcher/default.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/InazumaV/V2bX/common/counter"
 	"github.com/InazumaV/V2bX/common/rate"
 	"github.com/InazumaV/V2bX/limiter"
 
@@ -98,12 +99,13 @@ func (r *cachedReader) Interrupt() {
 
 // DefaultDispatcher is a default implementation of Dispatcher.
 type DefaultDispatcher struct {
-	ohm    outbound.Manager
+	ohm     outbound.Manager
-	router routing.Router
+	router  routing.Router
-	policy policy.Manager
+	policy  policy.Manager
-	stats  stats.Manager
+	stats   stats.Manager
-	fdns   dns.FakeDNSEngine
+	fdns    dns.FakeDNSEngine
-	Wm     *WriterManager
+	Wm      *WriterManager
+	Counter sync.Map
 }
 
 func init() {
@@ -204,24 +206,22 @@ func (d *DefaultDispatcher) getLink(ctx context.Context, network net.Network) (*
 			inboundLink.Writer = rate.NewRateLimitWriter(inboundLink.Writer, w)
 			outboundLink.Writer = rate.NewRateLimitWriter(outboundLink.Writer, w)
 		}
-		p := d.policy.ForLevel(user.Level)
+		var t *counter.TrafficCounter
-		if p.Stats.UserUplink {
+		if c, ok := d.Counter.Load(sessionInbound.Tag); !ok {
-			name := "user>>>" + user.Email + ">>>traffic>>>uplink"
+			t = counter.NewTrafficCounter()
-			if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
+			d.Counter.Store(sessionInbound.Tag, t)
-				inboundLink.Writer = &SizeStatWriter{
+		} else {
-					Counter: c,
+			t = c.(*counter.TrafficCounter)
-					Writer:  inboundLink.Writer,
-				}
-			}
 		}
-		if p.Stats.UserDownlink {
+
-			name := "user>>>" + user.Email + ">>>traffic>>>downlink"
+		inboundLink.Writer = &UploadTrafficWriter{
-			if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
+			Counter: t.GetCounter(user.Email),
-				outboundLink.Writer = &SizeStatWriter{
+			Writer:  inboundLink.Writer,
-					Counter: c,
+		}
-					Writer:  outboundLink.Writer,
+
-				}
+		outboundLink.Writer = &DownloadTrafficWriter{
-			}
+			Counter: t.GetCounter(user.Email),
+			Writer:  outboundLink.Writer,
 		}
 	}
 

core/xray/app/dispatcher/stats.go

@@ -1,25 +1,43 @@
 package dispatcher
 
 import (
+	"github.com/InazumaV/V2bX/common/counter"
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
-	"github.com/xtls/xray-core/features/stats"
 )
 
-type SizeStatWriter struct {
+type UploadTrafficWriter struct {
-	Counter stats.Counter
+	Counter *counter.TrafficStorage
 	Writer  buf.Writer
 }
 
-func (w *SizeStatWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+type DownloadTrafficWriter struct {
-	w.Counter.Add(int64(mb.Len()))
+	Counter *counter.TrafficStorage
+	Writer  buf.Writer
+}
+
+func (w *UploadTrafficWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	w.Counter.UpCounter.Add(int64(mb.Len()))
 	return w.Writer.WriteMultiBuffer(mb)
 }
 
-func (w *SizeStatWriter) Close() error {
+func (w *UploadTrafficWriter) Close() error {
 	return common.Close(w.Writer)
 }
 
-func (w *SizeStatWriter) Interrupt() {
+func (w *UploadTrafficWriter) Interrupt() {
+	common.Interrupt(w.Writer)
+}
+
+func (w *DownloadTrafficWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	w.Counter.DownCounter.Add(int64(mb.Len()))
+	return w.Writer.WriteMultiBuffer(mb)
+}
+
+func (w *DownloadTrafficWriter) Close() error {
+	return common.Close(w.Writer)
+}
+
+func (w *DownloadTrafficWriter) Interrupt() {
 	common.Interrupt(w.Writer)
 }

core/xray/node.go

@@ -17,6 +17,7 @@ type DNSConfig struct {
 }
 
 func (c *Xray) AddNode(tag string, info *panel.NodeInfo, config *conf.Options) error {
+	c.nodeReportMinTrafficBytes[tag] = config.ReportMinTraffic * 1024
 	err := updateDNSConfig(info)
 	if err != nil {
 		return fmt.Errorf("build dns error: %s", err)

core/xray/user.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/InazumaV/V2bX/api/panel"
+	"github.com/InazumaV/V2bX/common/counter"
 	"github.com/InazumaV/V2bX/common/format"
 	vCore "github.com/InazumaV/V2bX/core"
 	"github.com/xtls/xray-core/common/protocol"
@@ -32,47 +33,61 @@ func (c *Xray) DelUsers(users []panel.UserInfo, tag string, _ *panel.NodeInfo) e
 	if err != nil {
 		return fmt.Errorf("get user manager error: %s", err)
 	}
-	var up, down, user string
+	var user string
+	c.users.mapLock.Lock()
+	defer c.users.mapLock.Unlock()
 	for i := range users {
 		user = format.UserTag(tag, users[i].Uuid)
 		err = userManager.RemoveUser(context.Background(), user)
 		if err != nil {
 			return err
 		}
-		up = "user>>>" + user + ">>>traffic>>>uplink"
+		delete(c.users.uidMap, user)
-		down = "user>>>" + user + ">>>traffic>>>downlink"
+		c.dispatcher.Counter.Delete(user)
-		c.shm.UnregisterCounter(up)
-		c.shm.UnregisterCounter(down)
 		c.dispatcher.Wm.RemoveWritersForUser(user)
 	}
 	return nil
 }
 
-func (c *Xray) GetUserTraffic(tag, uuid string, reset bool) (up int64, down int64) {
+func (x *Xray) GetUserTrafficSlice(tag string, reset bool) ([]panel.UserTraffic, error) {
-	upName := "user>>>" + format.UserTag(tag, uuid) + ">>>traffic>>>uplink"
+	trafficSlice := make([]panel.UserTraffic, 0)
-	downName := "user>>>" + format.UserTag(tag, uuid) + ">>>traffic>>>downlink"
+	x.users.mapLock.RLock()
-	upCounter := c.shm.GetCounter(upName)
+	defer x.users.mapLock.RUnlock()
-	downCounter := c.shm.GetCounter(downName)
+	if v, ok := x.dispatcher.Counter.Load(tag); ok {
-	if reset {
+		c := v.(*counter.TrafficCounter)
-		if upCounter != nil {
+		c.Counters.Range(func(key, value interface{}) bool {
-			up = upCounter.Set(0)
+			email := key.(string)
-		}
+			traffic := value.(*counter.TrafficStorage)
-		if downCounter != nil {
+			up := traffic.UpCounter.Load()
-			down = downCounter.Set(0)
+			down := traffic.DownCounter.Load()
-		}
+			if up+down >= x.nodeReportMinTrafficBytes[tag] {
-	} else {
+				if reset {
-		if upCounter != nil {
+					traffic.UpCounter.Store(0)
-			up = upCounter.Value()
+					traffic.DownCounter.Store(0)
-		}
+				}
-		if downCounter != nil {
+				trafficSlice = append(trafficSlice, panel.UserTraffic{
-			down = downCounter.Value()
+					UID:      x.users.uidMap[email],
+					Upload:   up,
+					Download: down,
+				})
+			}
+			return true
+		})
+		if len(trafficSlice) == 0 {
+			return nil, nil
 		}
+		return trafficSlice, nil
 	}
-	return up, down
+	return nil, nil
 }
 
 func (c *Xray) AddUsers(p *vCore.AddUsersParams) (added int, err error) {
-	users := make([]*protocol.User, 0, len(p.Users))
+	c.users.mapLock.Lock()
+	defer c.users.mapLock.Unlock()
+	for i := range p.Users {
+		c.users.uidMap[format.UserTag(p.Tag, p.Users[i].Uuid)] = p.Users[i].Id
+	}
+	var users []*protocol.User
 	switch p.NodeInfo.Type {
 	case "vmess":
 		users = buildVmessUsers(p.Tag, p.Users)

core/xray/xray.go

@@ -30,16 +30,29 @@ func init() {
 
 // Xray Structure
 type Xray struct {
-	access     sync.Mutex
+	access                    sync.Mutex
-	Server     *core.Instance
+	Server                    *core.Instance
-	ihm        inbound.Manager
+	ihm                       inbound.Manager
-	ohm        outbound.Manager
+	ohm                       outbound.Manager
-	shm        statsFeature.Manager
+	shm                       statsFeature.Manager
-	dispatcher *dispatcher.DefaultDispatcher
+	dispatcher                *dispatcher.DefaultDispatcher
+	users                     *UserMap
+	nodeReportMinTrafficBytes map[string]int64
+}
+
+type UserMap struct {
+	uidMap  map[string]int
+	mapLock sync.RWMutex
 }
 
 func New(c *conf.CoreConfig) (vCore.Core, error) {
-	return &Xray{Server: getCore(c.XrayConfig)}, nil
+	return &Xray{
+		Server: getCore(c.XrayConfig),
+		users: &UserMap{
+			uidMap: make(map[string]int),
+		},
+		nodeReportMinTrafficBytes: make(map[string]int64),
+	}, nil
 }
 
 func parseConnectionConfig(c *conf.XrayConnectionConfig) (policy *coreConf.Policy) {

example/config.json

@@ -29,6 +29,7 @@
       "ListenIP": "0.0.0.0",
       "SendIP": "0.0.0.0",
       "DeviceOnlineMinTraffic": 200,
+      "MinReportTraffic": 0,
       "TCPFastOpen": false,
       "SniffEnabled": true,
       "CertConfig": {

node/user.go

@@ -8,20 +8,7 @@ import (
 )
 
 func (c *Controller) reportUserTrafficTask() (err error) {
-	// Get User traffic
+	userTraffic, err := c.server.GetUserTrafficSlice(c.tag, true)
-	userTraffic := make([]panel.UserTraffic, 0)
-	for i := range c.userList {
-		up, down := c.server.GetUserTraffic(c.tag, c.userList[i].Uuid, true)
-		if up > 0 || down > 0 {
-			if c.LimitConfig.EnableDynamicSpeedLimit {
-				c.traffic[c.userList[i].Uuid] += up + down
-			}
-			userTraffic = append(userTraffic, panel.UserTraffic{
-				UID:      (c.userList)[i].Id,
-				Upload:   up,
-				Download: down})
-		}
-	}
 	if len(userTraffic) > 0 {
 		err = c.apiClient.ReportUserTraffic(userTraffic)
 		if err != nil {