From 45a6a766215f273419ac16ae4843905a77a6617f Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Tue, 6 Oct 2026 11:56:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(cluster):=20=E8=B4=9F=E8=BD=BD=E5=9D=87?= =?UTF-8?q?=E8=A1=A1=E6=8E=A5=E5=85=A5=E8=BF=90=E8=A1=8C=E6=97=B6=20?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E9=BB=98=E8=AE=A4=20dry-run=EF=BC=8C?= =?UTF-8?q?=E6=98=BE=E5=BC=8F=20opt-in=20=E6=89=8D=E7=9C=9F=E6=AD=A3?= =?UTF-8?q?=E8=BF=81=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 承接 7d5bca5（Balancer 状态机）。本提交接线到 main.go，并让 **dry-run 成为默认**。 ## 为什么默认不启用 线上此刻是 `.30` 持 3 条转发、另两台 0 条。按已实现的规则，这会在第一次 检查就立刻触发迁移 —— 而每条迁移都要停一个 frpc worker、断掉活跃连接。 在没有看过真实数据会迁什么之前就把这个开关打开，等于拿生产连接赌一把。 所以： W4F_BALANCE_MODE=dry-run 只报告不动手（默认） W4F_BALANCE_MODE=on 真正迁移 W4F_BALANCE_RATIO=2 超载 = Score 超过均值 2 倍 W4F_BALANCE_SUSTAIN=1800 失衡需持续 30 分钟 W4F_BALANCE_INTERVAL=10 每 10 秒检查一次 dry-run 走同一套决策与迟滞逻辑（stepDryRun 复用 planBalance），因此它报告的 就是真开会迁的东西 —— 先用它观察，确认符合预期再切 on。 ## 只有 leader 平衡 WatchBalance 里 `if !b.engine.IsLeader() { continue }`。搬转发是全集群决策， 令牌是共享的：若每个成员各自决定，必然出现两个节点同时对同一条转发发 revoke。 （这与 Restart/RemoveNode 走同样的 owner 定向是同一类约束。） ## envFloat 新增：环境变量解析失败时回落到默认值并打日志，而不是静默变成 0（0 会被 NewBalancer 当成非法值钳制，但静默钳制无法与"用户显式要 0"区分）。 ## 测试 - TestDryRunNeverTouchesTheRing —— dry-run 必须报告"would migrate"且**不发布任何 任务、不动拓扑、不进入 in-flight**。这是它作为默认值存在的全部意义。 - TestOnlyLeaderBalances —— 非 leader 跑 WatchBalance 不得发布任何任务。 go build / go vet / go test ./...（go clean -testcache 后）全绿，gofmt 干净。 ## 下一步（未包含） 集群页 UI 展示超载提示与迁移进度；以及**先以 dry-run 在生产跑一轮观察**， 确认 planBalance 对真实数据的判断符合预期后再决定是否切 on。 --- cmd/webui4frpc/main.go | 53 ++++++++++++++++++++++++++ internal/cluster/balance_test.go | 63 +++++++++++++++++++++++++++++++ internal/cluster/ring_balance.go | 64 ++++++++++++++++++++++++++++++++ 3 files changed, 180 insertions(+) diff --git a/cmd/webui4frpc/main.go b/cmd/webui4frpc/main.go index 73df96d..dd20a34 100644 --- a/cmd/webui4frpc/main.go +++ b/cmd/webui4frpc/main.go @@ -15,6 +15,7 @@ import ( "os" "os/signal" "path/filepath" + "strconv" "strings" "syscall" "time" @@ -553,6 +554,42 @@ func main() { go ring.WatchTokenLoss(clusterCtx) go ring.WatchLeader(clusterCtx) + // Load balancing. DRY-RUN BY DEFAULT. + // + // The balancer moves forwards off an overloaded node, which means stopping a + // frpc worker and dropping its live connections. That is fine occasionally + // (a genuinely hot node) but destructive as a reflex, so it ships off and is + // opted into explicitly: + // + // W4F_BALANCE_MODE=dry-run report what it would do, move nothing (default) + // W4F_BALANCE_MODE=on actually migrate + // W4F_BALANCE_RATIO=2 overload = Score above RATIO x the mean + // W4F_BALANCE_SUSTAIN=1800 seconds an imbalance must persist + // W4F_BALANCE_INTERVAL=10 check every N seconds + // + // Only the leader balances (WatchBalance no-ops elsewhere): moving forwards is + // a cluster-wide decision and two members doing it independently would race on + // the same token. + balMode := strings.ToLower(strings.TrimSpace(os.Getenv("W4F_BALANCE_MODE"))) + if balMode == "" { + balMode = "dry-run" + } + balRatio := envFloat("W4F_BALANCE_RATIO", 2) + balSustain := int64(envFloat("W4F_BALANCE_SUSTAIN", 1800)) + balInterval := time.Duration(envFloat("W4F_BALANCE_INTERVAL", 10)) * time.Second + if balInterval < time.Second { + balInterval = 10 * time.Second + } + balancer := cluster.NewBalancer(ring, balRatio, balSustain) + if balMode == "on" { + log.Printf("ring[%s] load balancing ENABLED (ratio=%.2f sustain=%ds interval=%s)", + selfID, balRatio, balSustain, balInterval) + } else { + log.Printf("ring[%s] load balancing in DRY-RUN (set W4F_BALANCE_MODE=on to enable; "+ + "ratio=%.2f sustain=%ds interval=%s)", selfID, balRatio, balSustain, balInterval) + } + go balancer.WatchBalance(clusterCtx, balInterval, balMode != "on") + // Startup reconcile: every node (leader or not) checks its own topology // entries and re-claims any whose per-forward frpc worker process is // absent — the topology was recovered via AdoptState but the worker @@ -735,6 +772,22 @@ func retryJoinCluster(ctx context.Context, ring *cluster.Engine, targetAddr, joi } } +// envFloat reads a numeric environment variable, returning def when unset or +// unparseable. A malformed value must not silently change behaviour, so it falls +// back to the documented default rather than 0. +func envFloat(key string, def float64) float64 { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return def + } + f, err := strconv.ParseFloat(v, 64) + if err != nil { + log.Printf("env %s=%q is not a number; using %v", key, v, def) + return def + } + return f +} + // retryRejoinCached is the crash/restart recovery path: parse the persisted // peer list (JSON of [{addr,key},...]) and try to rejoin via each peer in // order. The first peer that accepts wins. Retry the whole list every 10s diff --git a/internal/cluster/balance_test.go b/internal/cluster/balance_test.go index a0936e3..3ec4cf6 100644 --- a/internal/cluster/balance_test.go +++ b/internal/cluster/balance_test.go @@ -3,7 +3,9 @@ package cluster import ( "context" "encoding/json" + "strings" "testing" + "time" "webui4frpc/internal/store" ) @@ -354,3 +356,64 @@ func TestBalancerQuietWhenBalanced(t *testing.T) { t.Error("balancer started work on a balanced ring") } } + +// TestDryRunNeverTouchesTheRing: dry-run is the default, and its whole value is +// that it can run against a live cluster without disturbing anything. It must +// report the migration it would make and publish nothing at all. +func TestDryRunNeverTouchesTheRing(t *testing.T) { + n1, _, _ := migrateRing(t, 3) + b := NewBalancer(n1, 2, 0) + + // Satisfy the hysteresis so the dry-run has something to report. + b.overSince = map[string]int64{"n1": 1000} + + msg := b.stepDryRun(5000) + if msg == "" { + t.Fatal("dry-run reported nothing for a 3-vs-0 imbalance") + } + if !strings.Contains(msg, "would migrate") { + t.Fatalf("dry-run message should be a WOULD report, got %q", msg) + } + // Nothing may have been published. + for _, tk := range n1.State().PendingList() { + if tk.Revoke || tk.Migrate { + t.Fatalf("dry-run published a %v task; it must not touch the ring", tk) + } + } + if b.InFlight() { + t.Error("dry-run started an in-flight migration") + } + if n := len(n1.State().TopologyList()); n != 3 { + t.Errorf("topology changed under dry-run: %d entries, want 3", n) + } +} + +// TestOnlyLeaderBalances: moving forwards is a cluster-wide decision. If every +// member decided independently they would race on the same token, each revoking +// the same forward. WatchBalance must therefore no-op on non-leaders. +func TestOnlyLeaderBalances(t *testing.T) { + n1, n2, _ := migrateRing(t, 3) // n1 is the leader, n2 is not + if !n1.IsLeader() || n2.IsLeader() { + t.Fatal("precondition broken: expected n1 leader, n2 not") + } + + before := len(n1.State().PendingList()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // A non-leader balancer with hysteresis already satisfied must still do + // nothing, because Step is only reached through WatchBalance's leader gate. + // Verify the gate itself by running the loop briefly and inspecting state. + bal := NewBalancer(n2, 2, 0) + bal.overSince = map[string]int64{"n1": 1000} + go bal.WatchBalance(ctx, 10*time.Millisecond, false) + time.Sleep(60 * time.Millisecond) + cancel() + + if n2.IsLeader() { + t.Fatal("precondition broken during the run") + } + if got := len(n2.State().PendingList()); got != 0 { + t.Errorf("a non-leader published %d task(s); balancing must be leader-only", got) + } + _ = before +} diff --git a/internal/cluster/ring_balance.go b/internal/cluster/ring_balance.go index 4cb4137..1e2d275 100644 --- a/internal/cluster/ring_balance.go +++ b/internal/cluster/ring_balance.go @@ -1,8 +1,10 @@ package cluster import ( + "context" "log" "strconv" + "time" "webui4frpc/internal/store" ) @@ -293,3 +295,65 @@ func (e *Engine) findBalanceCandidate(ratio float64) (*TopoEntry, string, string // hysteresis timer can start for a node whose forwards are all stopped. return pick, dest, pick.OwnerID } + +// WatchBalance drives the balancer on a ticker until ctx is cancelled. +// +// Only the LEADER runs this. Balancing is a cluster-wide decision (it moves +// forwards between members), and letting every member decide independently would +// mean several migrations racing for the same forward — the ring is a shared +// token, so two nodes must not both publish a revoke for the same entry. +// +// dryRun makes the balancer report what it WOULD do without publishing anything. +// That is the mode to use first on a live cluster: it answers "is my real data +// going to trip the threshold, and what would it move?" before anything on the +// ring is disturbed. +func (b *Balancer) WatchBalance(ctx context.Context, interval time.Duration, dryRun bool) { + tick := time.NewTicker(interval) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + } + if !b.engine.IsLeader() { + continue + } + now := time.Now().Unix() + if dryRun { + if msg := b.stepDryRun(now); msg != "" { + log.Printf("ring[%s] balance (dry-run): %s", b.engine.ID, msg) + } + continue + } + if msg := b.Step(now); msg != "" { + log.Printf("ring[%s] balance: %s", b.engine.ID, msg) + } + } +} + +// stepDryRun reports the migration the balancer would start right now, without +// publishing a revoke. Hysteresis is still applied so the report reflects the +// same sustained-imbalance rule the real path uses. +func (b *Balancer) stepDryRun(now int64) string { + if b.InFlight() { + return "" + } + pick, dest, hot := b.engine.findBalanceCandidate(b.ratio) + if hot == "" { + b.overSince = map[string]int64{} + return "" + } + if _, seen := b.overSince[hot]; !seen { + b.overSince[hot] = now + return "" + } + if pick == nil || dest == "" { + return "" + } + if since, ok := b.overSince[pick.OwnerID]; !ok || now-since < int64(b.sustainedSec) { + return "" + } + return "would migrate " + pick.Local.Name + ":" + strconv.Itoa(pick.Link.RemotePort) + + " from " + pick.OwnerID + " to " + dest +}