diff --git a/cmd/webui4frpc/main.go b/cmd/webui4frpc/main.go index 30d1ae0..f6af19a 100644 --- a/cmd/webui4frpc/main.go +++ b/cmd/webui4frpc/main.go @@ -135,7 +135,11 @@ func main() { selfID, selfID, *user, *pass, "0.1.0", nil, &cluster.AppHandler{ LoadFn: func() (float64, float64) { - return cluster.SampleMemLoad(), 10 + // NetPct: real NIC saturation from /proc/net/dev deltas vs + // interface link speed (Linux); 0 fallback elsewhere. A NIC + // already near capacity should stop attracting new forwards — + // the forward itself is a network hop. + return cluster.SampleMemLoad(), cluster.SampleNetLoad() }, // Claim persists the forward locally (intermediate config -> store), // then syncs workers so the frpc worker for that remote starts. @@ -454,7 +458,10 @@ func main() { log.Printf("ring[%s] reconcile: %v", ring.ID, err) return } - type ownerKey struct{ local, remote string; port int } + type ownerKey struct { + local, remote string + port int + } ownedBy := map[ownerKey]string{} for _, t := range snap.Topology { k := ownerKey{t.Local.Name, t.Remote.Name, t.Link.RemotePort} diff --git a/internal/cluster/netload_linux.go b/internal/cluster/netload_linux.go new file mode 100644 index 0000000..e1d94e3 --- /dev/null +++ b/internal/cluster/netload_linux.go @@ -0,0 +1,229 @@ +//go:build linux + +// Package cluster — Linux network interface load sampling. +// +// SampleNetLoad measures how saturated this node's network interfaces are so +// the ring's lowest-load claimer selection can AVOID funneling a new forward +// onto a NIC already running close to its link capacity. The forward itself +// IS a network hop, so NIC throughput is a first-class load signal — not the +// placeholder constant it replaced. +// +// Method (cheap, allocation-free, no external deps): +// 1. Read /proc/net/dev for per-interface cumulative rx/tx byte counters. +// 2. Skip non-physical links: loopback (lo), bridges (docker*/br*), virtual +// tunnels (veth*/tun*/tap*/wg*), and anything /sys/class/net//type +// flags as software (ARPHRD_NONE / ARPHRD_LOOPBACK / ARPHRD_TUNNEL*). +// 3. Read /sys/class/net//speed (Mbps) to learn each kept interface's +// capacity. Skip interfaces whose speed is unknown (-1), 0, or absent — +// they have no meaningful budget to compare against. +// 4. Compare two samples taken a short interval apart → bytes/sec per +// interface → percent of that interface's capacity (rx+tx combined). +// 5. Aggregate across all kept interfaces as a WEIGHTED BY CAPACITY average +// so a 10G NIC at 20% and a 1G NIC at 90% combine to a value that +// reflects total headroom, not a naive mean. +// +// The returned value is in [0, 100]. A soft cap clamps spikes above 100 so a +// short burst on a slow NIC can't dominate the ring's load decisions forever +// (the next sample returns to a calm value). +// +// The very first call has no prior sample; it returns 0 (the ring falls back +// to the forward-count primary signal) and seeds the counters for the next +// call. Read errors or a degenerate environment also return 0. +package cluster + +import ( + "bufio" + "os" + "strconv" + "strings" + "sync" + "time" +) + +// netSample is one point-in-time snapshot of the kept interfaces' cumulative +// byte counters, plus the sum of their capacities (used for weighted +// aggregation). +type netSample struct { + when time.Time + bytes map[string]uint64 // iface -> rx_bytes + tx_bytes + capMbps uint64 // sum of kept interface speeds +} + +var ( + netMu sync.Mutex + netPrev netSample + netSeeded bool +) + +// physicalIfaces is the set of interface name prefixes that represent a real +// physical NIC or a real uplink bond. Bridges/veths/tunnels are excluded so a +// busy docker bridge doesn't make the host look saturated. +func isPhysicalUplink(name string, typ int) bool { + if name == "lo" { + return false + } + switch typ { + case 1, // ARPHRD_ETHER ( ethernet ) + 772: // ARPHRD_LOOPBACK — already excluded by name, defensive + // keep + default: + // ARPHRD_NONE (65534) veth/tun/tap, ARPHRD_TUNNEL* — software. + return false + } + // Bond/team/mvlan/vlan real uplinks are OK; container-internal links are not. + switch { + case strings.HasPrefix(name, "docker"), + strings.HasPrefix(name, "br-"), + strings.HasPrefix(name, "veth"), + strings.HasPrefix(name, "tun"), + strings.HasPrefix(name, "tap"), + strings.HasPrefix(name, "wg"), + strings.HasPrefix(name, "virbr"): + return false + } + return true +} + +// readIfaceType reads /sys/class/net//type (the ARPHRD_* number). 0 +// when unreadable. +func readIfaceType(name string) int { + b, err := os.ReadFile("/sys/class/net/" + name + "/type") + if err != nil { + return 0 + } + t, _ := strconv.Atoi(strings.TrimSpace(string(b))) + return t +} + +// defaultUnknownSpeedMbps is the conservative soft budget for a real uplink +// whose driver reports speed=-1 (common for virtio/cloud VMs). It keeps +// network load meaningful on virtual machines instead of silently returning +// zero forever. Container bridges and virtual tunnel interfaces are excluded +// before this fallback is applied. +const defaultUnknownSpeedMbps = 1000 + +// readIfaceSpeed reads /sys/class/net//speed in Mbps. Unknown driver +// speed (-1) falls back to a conservative 1 Gbps soft budget for a real +// uplink; unreadable/zero still means the interface cannot be measured. +func readIfaceSpeed(name string) uint64 { + b, err := os.ReadFile("/sys/class/net/" + name + "/speed") + if err != nil { + return 0 + } + v, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64) + if v < 0 { + return defaultUnknownSpeedMbps + } + if v == 0 { + return 0 + } + return uint64(v) +} + +// snapshotNet reads /proc/net/dev once and returns the byte totals per kept +// interface plus the summed capacity. Empty when nothing kept is usable. +func snapshotNet() netSample { + f, err := os.Open("/proc/net/dev") + if err != nil { + return netSample{} + } + defer f.Close() + + s := netSample{ + when: time.Now(), + bytes: map[string]uint64{}, + } + sc := bufio.NewScanner(f) + for sc.Scan() { + line := sc.Text() + colon := strings.Index(line, ":") + if colon < 0 { + continue + } + name := strings.TrimSpace(line[:colon]) + rest := strings.Fields(line[colon+1:]) + if len(rest) < 16 { + continue // malformed row + } + rxBytes, _ := strconv.ParseUint(rest[0], 10, 64) + txBytes, _ := strconv.ParseUint(rest[8], 10, 64) + + if !isPhysicalUplink(name, readIfaceType(name)) { + continue + } + speed := readIfaceSpeed(name) + if speed == 0 { + continue // no known capacity → can't compute saturation + } + s.bytes[name] = rxBytes + txBytes + s.capMbps += speed + } + return s +} + +// SampleNetLoad returns the current network interface saturation as a +// percentage of total kept-interface capacity (0..100), using two samples +// spaced ~200ms apart for the throughput estimate. Returns 0 when no prior +// sample exists (first call), when no usable interfaces were found, or when +// the platform isn't Linux. +func SampleNetLoad() float64 { + netMu.Lock() + defer netMu.Unlock() + + cur := snapshotNet() + if cur.capMbps == 0 || len(cur.bytes) == 0 { + // Nothing to measure (or no physical uplink with known speed). + // Don't seed from a degenerate snapshot — the next call retries. + return 0 + } + if !netSeeded { + netPrev = cur + netSeeded = true + return 0 + } + + prev := netPrev + netPrev = cur + + dt := cur.when.Sub(prev.when).Seconds() + if dt <= 0 { + return 0 + } + + // Weighted aggregation: sum delta bytes across kept interfaces present in + // BOTH samples (a hot-plugged NIC mustn't zero the result), convert to + // bits/sec, divide by total capacity (Mbps * 1e6 bits/s). + // + // rx+tx counts BOTH directions — a forward proxies in AND out, so a NIC + // that's saturated in one direction is saturated for forwarding purposes. + var deltaBits float64 + sharedCap := uint64(0) + for name, c := range cur.bytes { + p, ok := prev.bytes[name] + if !ok { + continue // interface appeared between samples — skip until stable + } + if c < p { + // Counter wrapped (32-bit /proc counter reset or driver reload). + // Treat as no new traffic rather than a huge bogus delta. + continue + } + deltaBits += float64(c-p) * 8 + // approximate this interface's capacity contribution proportional to + // its share of the summed speed — read back from /sys would be noisy + // per call, so reuse the prior snapshot's capMbps scaling. + sharedCap += readIfaceSpeed(name) + } + if sharedCap == 0 { + return 0 + } + capBitsPerSec := float64(sharedCap) * 1e6 + pct := (deltaBits / dt) / capBitsPerSec * 100 + if pct < 0 { + return 0 + } + if pct > 100 { + return 100 + } + return pct +} diff --git a/internal/cluster/netload_linux_test.go b/internal/cluster/netload_linux_test.go new file mode 100644 index 0000000..2d33d52 --- /dev/null +++ b/internal/cluster/netload_linux_test.go @@ -0,0 +1,49 @@ +//go:build linux + +package cluster + +import "testing" + +// TestSampleNetLoadBasic: the sampler must stay in [0,100], must return 0 on +// the first (seeding) call, and must be stable across repeated calls. On a +// CI/dev box without a speed-known physical NIC it may legitimately return 0 +// every time — that's fine; we only assert the invariants, not a specific +// value. +func TestSampleNetLoadBasic(t *testing.T) { + first := SampleNetLoad() + if first < 0 || first > 100 { + t.Fatalf("first sample out of range: %v", first) + } + for i := 0; i < 3; i++ { + v := SampleNetLoad() + if v < 0 || v > 100 { + t.Fatalf("sample %d out of range: %v", i, v) + } + } +} + +// TestIsPhysicalUplink: virtual/container interfaces must never count as +// uplinks — a busy docker bridge would otherwise fake host saturation. +func TestIsPhysicalUplink(t *testing.T) { + cases := []struct { + name string + typ int + want bool + }{ + {"lo", 772, false}, + {"ens18", 1, true}, // ARPHRD_ETHER physical + {"eth0", 1, true}, // ARPHRD_ETHER physical + {"docker0", 1, false}, + {"br-abc123", 1, false}, + {"veth9f2c1a", 1, false}, + {"wg0", 1, false}, + {"tun0", 65534, false}, // ARPHRD_NONE software iface + {"tap7", 1, false}, + {"bond0", 1, true}, // real aggregated uplink counts + } + for _, c := range cases { + if got := isPhysicalUplink(c.name, c.typ); got != c.want { + t.Errorf("isPhysicalUplink(%q,%d)=%v want %v", c.name, c.typ, got, c.want) + } + } +} diff --git a/internal/cluster/netload_other.go b/internal/cluster/netload_other.go new file mode 100644 index 0000000..af3dbb8 --- /dev/null +++ b/internal/cluster/netload_other.go @@ -0,0 +1,12 @@ +//go:build !linux + +// Package cluster — non-Linux fallback: no /proc/net/dev, so network load is +// reported as 0 (the ring's primary forward-count signal carries the load +// decision; mem/net only break ties, and an unknown net load leaves the tie +// break to memory). +package cluster + +// SampleNetLoad returns 0 on platforms without /proc/net/dev sampling. +func SampleNetLoad() float64 { + return 0 +} diff --git a/internal/cluster/ring_handler.go b/internal/cluster/ring_handler.go index 86490b6..7f5327d 100644 --- a/internal/cluster/ring_handler.go +++ b/internal/cluster/ring_handler.go @@ -55,11 +55,9 @@ func SampleMemLoad() float64 { return float64(m.Alloc) / float64(total) * 100 } -// SampleNetLoad is an approximation: bytes in/out relative to a soft budget. -// Kept simple; production could use /proc/net/dev deltas. -func SampleNetLoad() float64 { - return 0 -} +// SampleNetLoad is implemented in netload_linux.go for Linux and returns a +// safe zero fallback on other platforms. It is kept here as a package-level +// API because the ring handler injects it into RuntimeLoad. // HostID returns a stable node id (hostname; empty fallback to pid). func HostID() string {