mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-16 15:17:14 +00:00
Sanaei
56bb876d8d
A master-side network blip flips every node in one heartbeat tick, and each node published its own node.down, then node.up. A notifier queue holds 64 events and the rate limiter keys on the node name, so with 150 nodes most alerts were dropped and the rest ran into Telegram and Discord limits. Past five same-direction transitions in one tick the heartbeat publishes a single event per direction naming the nodes (the first ten, sorted, then +N). Smaller ticks keep per-node events with their health data, and the notifiers already read the node name from Source, so no formatter changed.
104 lines
3.1 KiB
Go
104 lines
3.1 KiB
Go
package job
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http/httptest"
|
|
"path/filepath"
|
|
"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"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/eventbus"
|
|
xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
|
)
|
|
|
|
// goingDownNodes seeds n online nodes whose address refuses connections, so the
|
|
// next heartbeat flips every one of them to offline in the same tick.
|
|
func goingDownNodes(t *testing.T, n int) {
|
|
t.Helper()
|
|
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() })
|
|
runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}}))
|
|
t.Cleanup(func() { runtime.SetManager(nil) })
|
|
srv := httptest.NewServer(nil)
|
|
host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
|
|
portNum, _ := strconv.Atoi(port)
|
|
srv.Close()
|
|
for i := range n {
|
|
node := &model.Node{
|
|
Name: fmt.Sprintf("node-%02d", i), Scheme: "http", Address: host, Port: portNum, BasePath: "/",
|
|
ApiToken: "tok", Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify",
|
|
}
|
|
if err := database.GetDB().Create(node).Error; err != nil {
|
|
t.Fatalf("create node: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func collectNodeEvents(t *testing.T) func() []eventbus.Event {
|
|
t.Helper()
|
|
bus := eventbus.New(eventbus.DefaultBufferSize)
|
|
var mu sync.Mutex
|
|
var got []eventbus.Event
|
|
bus.Subscribe("test", func(e eventbus.Event) {
|
|
mu.Lock()
|
|
got = append(got, e)
|
|
mu.Unlock()
|
|
})
|
|
prev := EventBus
|
|
EventBus = bus
|
|
t.Cleanup(func() {
|
|
EventBus = prev
|
|
bus.Stop()
|
|
})
|
|
return func() []eventbus.Event {
|
|
time.Sleep(300 * time.Millisecond)
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
return append([]eventbus.Event(nil), got...)
|
|
}
|
|
}
|
|
|
|
// A master-side blip flipped every node in one tick and published one event per
|
|
// node, overflowing the notifier queues and every chat's rate limit.
|
|
func TestHeartbeatSummarizesNodeDownBurst(t *testing.T) {
|
|
goingDownNodes(t, 12)
|
|
events := collectNodeEvents(t)
|
|
|
|
NewNodeHeartbeatJob().Run()
|
|
|
|
got := events()
|
|
if len(got) != 1 {
|
|
t.Fatalf("heartbeat published %d events for 12 nodes going down, want 1 summary", len(got))
|
|
}
|
|
want := "node-00, node-01, node-02, node-03, node-04, node-05, node-06, node-07, node-08, node-09 (+2)"
|
|
if got[0].Type != eventbus.EventNodeDown || got[0].Source != want {
|
|
t.Fatalf("summary event = %s %q, want %s %q", got[0].Type, got[0].Source, eventbus.EventNodeDown, want)
|
|
}
|
|
}
|
|
|
|
func TestHeartbeatKeepsSingleNodeDownEvent(t *testing.T) {
|
|
goingDownNodes(t, 1)
|
|
events := collectNodeEvents(t)
|
|
|
|
NewNodeHeartbeatJob().Run()
|
|
|
|
got := events()
|
|
if len(got) != 1 || got[0].Source != "node-00" {
|
|
t.Fatalf("events = %+v, want the node's own node.down", got)
|
|
}
|
|
if _, ok := got[0].Data.(*eventbus.NodeHealthData); !ok {
|
|
t.Fatalf("single node.down lost its health data: %#v", got[0].Data)
|
|
}
|
|
}
|