mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-17 15:47:14 +00:00
fix(node): push a node only the client IPs it hosts
A master's per-node sync must scope what it sends to the clients that node serves, so its cost tracks the node and not the fleet. The global-usage push already did (node_client_traffics by node_id); the 10s client-IP push sent GetAllInboundClientIps, the whole table, to every node. Each node's MergeInboundClientIps then created a row for every foreign email, and its next GET clientIps echoed the whole fleet back. Its IP-limit job only ever reads rows for its own clients, so none of it was used. With 150 nodes x 150 clients, one IP tick pushed 299 MB and pulled 264 MB, every node held 22,500 rows instead of 150, and sync ticks grew 3.8s -> 10.2s even at 1ms latency; the cost grows with the square of the fleet. Both pushes now share nodeHostedEmails. After the change the same fleet moves 2.0 MB / 1.8 MB per tick and ticks stay near 3.2s. Nodes upgraded with foreign rows shed them within 30 minutes via pruneStaleIpRows.
This commit is contained in:
@@ -0,0 +1,88 @@
|
|||||||
|
package job
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"path/filepath"
|
||||||
|
"slices"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/op/go-logging"
|
||||||
|
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||||
|
xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
||||||
|
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A node's IP-limit job only reads rows for its own clients, so pushing the
|
||||||
|
// whole table made every node store and echo back the entire fleet's IPs.
|
||||||
|
func TestNodeTrafficSyncPushesOnlyHostedClientIps(t *testing.T) {
|
||||||
|
xuilogger.InitLogger(logging.ERROR)
|
||||||
|
if err := database.InitDB(filepath.Join(t.TempDir(), "x-ui.db")); err != nil {
|
||||||
|
t.Fatalf("InitDB: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = database.CloseDB() })
|
||||||
|
service.StartTrafficWriter()
|
||||||
|
t.Cleanup(service.StopTrafficWriter)
|
||||||
|
runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
|
||||||
|
t.Cleanup(func() { runtime.SetManager(nil) })
|
||||||
|
|
||||||
|
var mu sync.Mutex
|
||||||
|
pushed := map[string][]string{}
|
||||||
|
now := time.Now().Unix()
|
||||||
|
for i, email := range []string{"a@node", "b@node"} {
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
switch {
|
||||||
|
case strings.HasSuffix(r.URL.Path, "inbounds/list"):
|
||||||
|
settings := fmt.Sprintf(`{"clients":[{"email":%q,"id":"0000000%d-0000-4000-8000-000000000000","enable":true}],"decryption":"none"}`, email, i)
|
||||||
|
ib, _ := json.Marshal([]map[string]any{{
|
||||||
|
"id": 1, "tag": fmt.Sprintf("in-%d", 20000+i), "port": 20000 + i, "protocol": "vless", "enable": true,
|
||||||
|
"settings": settings, "streamSettings": `{"network":"tcp"}`, "sniffing": `{}`,
|
||||||
|
"clientStats": []map[string]any{{"email": email, "enable": true}},
|
||||||
|
}})
|
||||||
|
_, _ = w.Write([]byte(`{"success":true,"obj":` + string(ib) + `}`))
|
||||||
|
return
|
||||||
|
case strings.HasSuffix(r.URL.Path, "server/clientIps") && r.Method == http.MethodPost:
|
||||||
|
var rows []model.InboundClientIps
|
||||||
|
_ = json.NewDecoder(r.Body).Decode(&rows)
|
||||||
|
mu.Lock()
|
||||||
|
for _, row := range rows {
|
||||||
|
pushed[email] = append(pushed[email], row.ClientEmail)
|
||||||
|
}
|
||||||
|
mu.Unlock()
|
||||||
|
}
|
||||||
|
_, _ = w.Write([]byte(`{"success":true}`))
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
|
||||||
|
portNum, _ := strconv.Atoi(port)
|
||||||
|
if err := database.GetDB().Create(&model.Node{
|
||||||
|
Name: email, Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok",
|
||||||
|
Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify",
|
||||||
|
}).Error; err != nil {
|
||||||
|
t.Fatalf("create node: %v", err)
|
||||||
|
}
|
||||||
|
if err := database.GetDB().Create(&model.InboundClientIps{
|
||||||
|
ClientEmail: email, Ips: fmt.Sprintf(`[{"ip":"10.0.0.%d","timestamp":%d}]`, i+1, now),
|
||||||
|
}).Error; err != nil {
|
||||||
|
t.Fatalf("seed client ips: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
NewNodeTrafficSyncJob().Run()
|
||||||
|
|
||||||
|
for _, email := range []string{"a@node", "b@node"} {
|
||||||
|
if got := pushed[email]; !slices.Equal(got, []string{email}) {
|
||||||
|
t.Errorf("node hosting %s received IP rows for %v, want only [%s]", email, got, email)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -447,7 +447,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy
|
|||||||
logger.Warningf("node traffic sync: fetch client ips from %s failed: %v", n.Name, err)
|
logger.Warningf("node traffic sync: fetch client ips from %s failed: %v", n.Name, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
masterIps, err := j.inboundService.GetAllInboundClientIps()
|
masterIps, err := j.inboundService.GetNodeInboundClientIps(n.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Warningf("node traffic sync: load client ips for push to %s failed: %v", n.Name, err)
|
logger.Warningf("node traffic sync: load client ips for push to %s failed: %v", n.Name, err)
|
||||||
return active
|
return active
|
||||||
|
|||||||
@@ -19,6 +19,33 @@ func (s *InboundService) GetAllInboundClientIps() ([]model.InboundClientIps, err
|
|||||||
return ips, err
|
return ips, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// nodeHostedEmails is every client one node serves, its descendants' included.
|
||||||
|
// Per-node pushes are scoped to it so their cost tracks the node, not the fleet.
|
||||||
|
func nodeHostedEmails(db *gorm.DB, nodeID int) ([]string, error) {
|
||||||
|
var emails []string
|
||||||
|
err := db.Model(&model.NodeClientTraffic{}).Where("node_id = ?", nodeID).Pluck("email", &emails).Error
|
||||||
|
return emails, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetNodeInboundClientIps returns the IP rows of the clients nodeID hosts: a node's
|
||||||
|
// IP-limit job reads no other row, so pushing the rest only made it echo them back.
|
||||||
|
func (s *InboundService) GetNodeInboundClientIps(nodeID int) ([]model.InboundClientIps, error) {
|
||||||
|
db := database.GetDB()
|
||||||
|
emails, err := nodeHostedEmails(db, nodeID)
|
||||||
|
if err != nil || len(emails) == 0 {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var ips []model.InboundClientIps
|
||||||
|
for _, batch := range chunkStrings(emails, sqlInChunk) {
|
||||||
|
var page []model.InboundClientIps
|
||||||
|
if err := db.Where("client_email IN ?", batch).Find(&page).Error; err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
ips = append(ips, page...)
|
||||||
|
}
|
||||||
|
return ips, nil
|
||||||
|
}
|
||||||
|
|
||||||
// clientIpStaleAfterSeconds mirrors job.ipStaleAfterSeconds: client IPs older than
|
// clientIpStaleAfterSeconds mirrors job.ipStaleAfterSeconds: client IPs older than
|
||||||
// 30 minutes are evicted. Applying the same cutoff inside the cross-node merge keeps
|
// 30 minutes are evicted. Applying the same cutoff inside the cross-node merge keeps
|
||||||
// the synced blob bounded and stops the master's push-back from resurrecting IPs that
|
// the synced blob bounded and stops the master's push-back from resurrecting IPs that
|
||||||
|
|||||||
@@ -168,10 +168,8 @@ func overlayGlobalTrafficValues(db *gorm.DB, rows []xray.ClientTraffic) {
|
|||||||
// its own aggregate.
|
// its own aggregate.
|
||||||
func (s *InboundService) GetNodeClientTraffics(nodeID int) ([]*xray.ClientTraffic, error) {
|
func (s *InboundService) GetNodeClientTraffics(nodeID int) ([]*xray.ClientTraffic, error) {
|
||||||
db := database.GetDB()
|
db := database.GetDB()
|
||||||
var emails []string
|
emails, err := nodeHostedEmails(db, nodeID)
|
||||||
if err := db.Model(&model.NodeClientTraffic{}).
|
if err != nil {
|
||||||
Where("node_id = ?", nodeID).
|
|
||||||
Pluck("email", &emails).Error; err != nil {
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(emails) == 0 {
|
if len(emails) == 0 {
|
||||||
|
|||||||
Reference in New Issue
Block a user