diff --git a/cmd/webui4frpc/main.go b/cmd/webui4frpc/main.go index dd20a34..6745449 100644 --- a/cmd/webui4frpc/main.go +++ b/cmd/webui4frpc/main.go @@ -389,6 +389,11 @@ func main() { } }) + // Load balancer is created here (before the Handler) so the HTTP layer can + // expose its read-only report. Mode/parameters are applied further down, once + // the W4F_BALANCE_* environment has been parsed. + balancer := cluster.NewBalancer(ring, 2, 1800) + h := &httpapi.Handler{ Store: st, Process: pm, @@ -580,7 +585,7 @@ func main() { if balInterval < time.Second { balInterval = 10 * time.Second } - balancer := cluster.NewBalancer(ring, balRatio, balSustain) + balancer.SetRatioAndSustain(balRatio, balSustain) if balMode == "on" { log.Printf("ring[%s] load balancing ENABLED (ratio=%.2f sustain=%ds interval=%s)", selfID, balRatio, balSustain, balInterval) diff --git a/internal/cluster/balance_test.go b/internal/cluster/balance_test.go index 3ec4cf6..a5c9236 100644 --- a/internal/cluster/balance_test.go +++ b/internal/cluster/balance_test.go @@ -417,3 +417,107 @@ func TestOnlyLeaderBalances(t *testing.T) { } _ = before } + +// TestReportDoesNotDisturbHysteresis: Report is a read-only view that an operator +// can call at will. If it touched the overSince timer, a curious operator +// polling the endpoint would keep resetting the sustained-imbalance clock and +// balancing would never fire. It must observe without perturbing. +func TestReportDoesNotDisturbHysteresis(t *testing.T) { + n1, _, _ := migrateRing(t, 3) + b := NewBalancer(n1, 2, 1800) + + // First Step only starts the timer (no action yet). + _ = b.Step(1000) + started, seen := b.overSince["n1"] + if !seen { + t.Fatal("precondition: the overload timer should have been started") + } + + // Poll the report repeatedly, as an operator would. + for i := 0; i < 5; i++ { + rep := b.Report(int64(2000 + i*600)) + if len(rep.Nodes) != 3 { + t.Fatalf("report listed %d nodes, want 3", len(rep.Nodes)) + } + } + if got := b.overSince["n1"]; got != started { + t.Errorf("Report reset the overload timer: %d -> %d", started, got) + } + + // And the real path must still work: past the window it acts. + _ = b.Step(1000 + 1801) + if b.revoking == nil { + t.Error("polling the report suppressed the migration it should have allowed") + } +} + +// TestReportOnFollowerHasNoPlan: only the leader acts, so a follower must not +// claim it would migrate anything — otherwise an operator querying a random node +// gets an answer that does not reflect what the cluster will actually do. +func TestReportOnFollowerHasNoPlan(t *testing.T) { + _, n2, _ := migrateRing(t, 3) // n2 is not the leader + b := NewBalancer(n2, 2, 0) + b.overSince = map[string]int64{"n1": 1000} + + rep := b.Report(9999) + if rep.IsLeader { + t.Fatal("precondition: n2 must not be the leader") + } + if rep.WouldMigrate != nil { + t.Errorf("a follower reported a migration plan: %+v", rep.WouldMigrate) + } + if len(rep.Nodes) == 0 { + t.Error("a follower should still report the node scores (that part is informational)") + } +} + +// TestReportShowsPlanOnceHysteresisSatisfied pins the operator-facing contract: +// before the window elapses there is no plan; after it, the plan appears with +// the concrete forward and both endpoints. +func TestReportShowsPlanOnceHysteresisSatisfied(t *testing.T) { + n1, _, _ := migrateRing(t, 3) + b := NewBalancer(n1, 2, 1800) + + // Start the timer but do not satisfy the window. + b.overSince = map[string]int64{"n1": 1000} + if rep := b.Report(1500); rep.WouldMigrate != nil { + t.Errorf("plan reported before the sustained window elapsed: %+v", rep.WouldMigrate) + } + + rep := b.Report(1000 + 1801) + if rep.WouldMigrate == nil { + t.Fatal("no plan reported even though the imbalance is past the window") + } + m := rep.WouldMigrate + if m.From == "" || m.To == "" || m.Local == "" || m.Port == 0 { + t.Fatalf("plan is incomplete: %+v", m) + } + if m.From == m.To { + t.Fatalf("plan moves a forward to the node it already lives on: %+v", m) + } + if !rep.Enabled == false && rep.DryRun { + // sanity: default is observe-only + if !rep.DryRun { + t.Error("balancer should default to dry-run") + } + } +} + +// TestReportSurfacesInFlightMigration: the two-phase machine must be visible while +// it runs, otherwise an operator sees a stalled plan and cannot tell a slow revoke +// from a stuck one. +func TestReportSurfacesInFlightMigration(t *testing.T) { + n1, _, _ := migrateRing(t, 3) + b := NewBalancer(n1, 2, 0) + _ = b.Step(1000) // starts phase 1 + if !b.InFlight() { + t.Fatal("precondition: a migration should be in flight") + } + rep := b.Report(1001) + if rep.InFlight == nil { + t.Fatal("report did not surface the in-flight migration") + } + if rep.InFlight.Local == "" || rep.InFlight.To == "" { + t.Fatalf("in-flight plan is incomplete: %+v", rep.InFlight) + } +} diff --git a/internal/cluster/ring_balance.go b/internal/cluster/ring_balance.go index 1e2d275..844ca5e 100644 --- a/internal/cluster/ring_balance.go +++ b/internal/cluster/ring_balance.go @@ -154,6 +154,9 @@ type Balancer struct { // what stops a transient CPU/network spike from restarting workers. sustainedSec int64 + // dryRun records the mode set at construction so Report can surface it. + // SetDryRun updates it; main.go sets it once from W4F_BALANCE_MODE. + dryRun bool // overSince records when the currently-overloaded node first crossed the // threshold. Zero means "not currently over". overSince map[string]int64 @@ -183,10 +186,17 @@ func NewBalancer(e *Engine, ratio float64, sustainedSec int64) *Balancer { } return &Balancer{ engine: e, ratio: ratio, sustainedSec: sustainedSec, + dryRun: true, // safe default: report, never move overSince: map[string]int64{}, } } +// SetDryRun selects the mode. Dry-run (the default) only reports. +func (b *Balancer) SetDryRun(v bool) { b.dryRun = v } + +// DryRun reports whether the balancer is in observe-only mode. +func (b *Balancer) DryRun() bool { return b.dryRun } + // InFlight reports whether a migration is currently between phases. func (b *Balancer) InFlight() bool { return b.revoking != nil || b.publishing != nil @@ -357,3 +367,107 @@ func (b *Balancer) stepDryRun(now int64) string { return "would migrate " + pick.Local.Name + ":" + strconv.Itoa(pick.Link.RemotePort) + " from " + pick.OwnerID + " to " + dest } + +// BalanceReport is a read-only view of what the balancer sees and would do. +// +// It exists because the balancer's decision is otherwise only visible in the +// log, and only when an imbalance exists AND the hysteresis window has elapsed. +// An operator asking "is this cluster balanced?" gets silence in exactly the +// healthy case, which is indistinguishable from "balancing is switched off". +type BalanceReport struct { + Enabled bool `json:"enabled"` + DryRun bool `json:"dryRun"` + Ratio float64 `json:"ratio"` + SustainSec int64 `json:"sustainSec"` + // Only the leader acts; on a follower these stay false/empty. + IsLeader bool `json:"isLeader"` + // Current per-node scores. + Nodes []BalanceNodeReport `json:"nodes"` + // Overloaded (above ratio x mean) and sustained past the window, if any. + OverloadedSince int64 `json:"overloadedSince,omitempty"` + // The migration that would happen on the next Step, if any. + WouldMigrate *BalanceMigrationPlan `json:"wouldMigrate,omitempty"` + // Set when a migration is between phases. + InFlight *BalanceMigrationPlan `json:"inFlight,omitempty"` +} + +type BalanceNodeReport struct { + ID string `json:"id"` + Alive bool `json:"alive"` + IsLeader bool `json:"isLeader"` + Forwards int `json:"forwards"` + MemPct float64 `json:"memPct"` + NetPct float64 `json:"netPct"` + Score float64 `json:"score"` +} + +type BalanceMigrationPlan struct { + Local string `json:"local"` + Remote string `json:"remote"` + Port int `json:"remotePort"` + From string `json:"from"` + To string `json:"to"` +} + +// Report renders the balancer's current view. It never mutates anything — +// note that it does NOT touch overSince, so calling it cannot start or reset +// the hysteresis timer that the real Step path uses. +func (b *Balancer) Report(now int64) BalanceReport { + rep := BalanceReport{ + Enabled: !b.dryRun, + DryRun: b.dryRun, + Ratio: b.ratio, + SustainSec: int64(b.sustainedSec), + IsLeader: b.engine.IsLeader(), + } + for i := range b.engine.state.Nodes { + n := &b.engine.state.Nodes[i] + rep.Nodes = append(rep.Nodes, BalanceNodeReport{ + ID: n.ID, Alive: n.Alive, IsLeader: n.IsLeader, + Forwards: n.Load.Forwards, MemPct: n.Load.MemPct, NetPct: n.Load.NetPct, + Score: n.Load.Score(), + }) + } + if m := b.revoking; m != nil { + rep.InFlight = &BalanceMigrationPlan{ + Local: m.local, Remote: m.remote, Port: m.port, From: m.fromOwner, To: m.toOwner, + } + } + if m := b.publishing; m != nil { + rep.InFlight = &BalanceMigrationPlan{ + Local: m.local, Remote: m.remote, Port: m.port, From: m.fromOwner, To: m.toOwner, + } + } + // Only a leader would act, so only report a plan there. + if rep.IsLeader { + pick, dest, hot := b.engine.findBalanceCandidate(b.ratio) + if hot != "" { + rep.OverloadedSince = b.overSince[hot] + } + if pick != nil && dest != "" { + plan := &BalanceMigrationPlan{ + Local: pick.Local.Name, Remote: pick.Remote.Name, + Port: pick.Link.RemotePort, From: pick.OwnerID, To: dest, + } + // surface it as "would migrate" only once the window is satisfied + if since, ok := b.overSince[pick.OwnerID]; ok && now-since >= int64(b.sustainedSec) { + rep.WouldMigrate = plan + } + } + } + return rep +} + +// SetRatioAndSustain updates the thresholds after construction, so main.go can +// create the balancer early (the HTTP handler needs it) and apply the environment +// configuration once it has parsed it. +func (b *Balancer) SetRatioAndSustain(ratio float64, sustainedSec int64) { + if ratio <= 1 { + ratio = 2 + } + if sustainedSec < 0 { + sustainedSec = 0 + } + b.ratio = ratio + b.sustainedSec = sustainedSec +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 403fd00..ffe6a72 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -48,6 +48,8 @@ type Handler struct { Cluster *cluster.Registry // Ring is the token-ring engine (M6). Nil disables ring routes. Ring *cluster.Engine + // Balancer is optional: when nil the balance endpoint reports 404. + Balancer *cluster.Balancer // SelfAddr is this node's reachable listen address (from -addr), used for // cluster discovery so peers can reach back for binary exchange. SelfAddr string @@ -114,6 +116,7 @@ func NewServeMux(h *Handler) (http.Handler, error) { mux.HandleFunc(apiPrefix+"/cluster/nodes", h.auth("read")(h.handleClusterNodes)) mux.HandleFunc(apiPrefix+"/cluster/cache", h.auth("read")(h.handleClusterCache)) // GET read; POST write-checked inside mux.HandleFunc(apiPrefix+"/cluster/ring", h.auth("read")(h.handleClusterRing)) + mux.HandleFunc(apiPrefix+"/cluster/balance", h.auth("read")(h.handleClusterBalance)) mux.HandleFunc(apiPrefix+"/node/logs", h.auth("read")(h.handleNodeLogs)) mux.HandleFunc(apiPrefix+"/cluster/logs/export", h.auth("read")(h.handleClusterLogsExport)) @@ -634,6 +637,27 @@ func (h *Handler) handleClusterRing(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, h.Ring.Snapshot()) } +// handleClusterBalance reports the load balancer's read-only view: per-node +// scores, whether anything is over the threshold, and the migration it would +// perform. Publishing nothing is the point — an operator asking "is this cluster +// balanced?" currently gets silence from the log in the healthy case, which is +// indistinguishable from balancing being switched off. +func (h *Handler) handleClusterBalance(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + methodNotAllowed(w) + return + } + if h.Ring == nil { + http.Error(w, "ring engine not enabled", http.StatusNotFound) + return + } + if h.Balancer == nil { + http.Error(w, "load balancer not enabled", http.StatusNotFound) + return + } + writeJSON(w, http.StatusOK, h.Balancer.Report(time.Now().Unix())) +} + // handleClusterJoin accepts a newcomer join request: the target node inserts // the newcomer after itself in the ring and returns the updated ring state // for the newcomer to adopt.