//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. // Capacity comes from the CURRENT snapshot's capMbps (already summed in // snapshotNet) — no extra /sys reads per interface here. var deltaBits float64 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 } if cur.capMbps == 0 { return 0 } capBitsPerSec := float64(cur.capMbps) * 1e6 pct := (deltaBits / dt) / capBitsPerSec * 100 if pct < 0 { return 0 } if pct > 100 { return 100 } return pct }