package monitor import ( "context" crand "crypto/rand" "math" "math/big" "strings" "sync" "time" ) // Probe scheduling and change-detection thresholds. const ( jitterBaseMillis = 100 jitterRangeMillis = 800 lossChangeThreshold = 0.01 tcpChangeThreshold = 20 // milliseconds ) // randomOffset returns a startup jitter to avoid clustering probes at the // same instant across loops. crypto/rand is used so no weak PRNG is linked. func randomOffset() time.Duration { n, err := crand.Int(crand.Reader, big.NewInt(jitterRangeMillis)) if err != nil { return jitterBaseMillis * time.Millisecond } return time.Duration(jitterBaseMillis+n.Int64()) * time.Millisecond } // probeAll runs probe for each host concurrently and returns the results // keyed by host. func probeAll[T any](hosts []string, probe func(host string) T) map[string]T { var ( wg sync.WaitGroup mu sync.Mutex res = make(map[string]T, len(hosts)) ) for _, h := range hosts { wg.Add(1) go func(host string) { defer wg.Done() v := probe(host) mu.Lock() res[host] = v mu.Unlock() }(h) } wg.Wait() return res } // reachLoop monitors reachability for hosts on one interface. func (m *Monitor) reachLoop(ctx context.Context, st *InterfaceStatus, hosts []string) { m.logf("Starting reachability monitoring for %s with %d hosts", st.Name, len(hosts)) time.Sleep(randomOffset()) tk := time.NewTicker(time.Second) defer tk.Stop() for { select { case <-ctx.Done(): m.logf("Stopping reachability monitoring for %s", st.Name) return case <-tk.C: m.reachTick(ctx, st, hosts) } } } // reachTick pings every host concurrently and applies the results. func (m *Monitor) reachTick(ctx context.Context, st *InterfaceStatus, hosts []string) { res := probeAll(hosts, func(host string) bool { return m.reachProbe(ctx, st, host) }) m.reachApply(st, res) } // reachProbe pings a single host and updates per-host counters. func (m *Monitor) reachProbe( ctx context.Context, st *InterfaceStatus, host string, ) bool { st.mu.Lock() st.TotalICMPReq++ st.MeterValue = min(st.MeterValue+1, m.MaxMeterValue) st.mu.Unlock() ok := m.pingOnce(ctx, st.Name, host) st.mu.Lock() defer st.mu.Unlock() if ok { st.TotalICMPRep++ st.MeterValue = max(st.MeterValue-1, 0) st.Spin() return true } st.DroppedCount++ st.LastDrop = time.Now() st.LostPackets[host]++ m.notifyUI() return false } // reachApply stores the round's results and decays the meter when clean. func (m *Monitor) reachApply(st *InterfaceStatus, res map[string]bool) { st.mu.Lock() changed := reachChanged(st.Reachable, res) st.Reachable = res st.LastPing = time.Now() if allReachable(res) && st.MeterValue > 0 { st.MeterValue-- } st.mu.Unlock() if changed { m.notifyUI() } } // reachChanged reports whether any host's reachability differs from before. func reachChanged(old, cur map[string]bool) bool { for host, newStatus := range cur { if oldStatus, ok := old[host]; !ok || oldStatus != newStatus { return true } } return false } // allReachable reports whether every host in the map is reachable. func allReachable(res map[string]bool) bool { for _, ok := range res { if !ok { return false } } return true } // lossLoop monitors packet loss for hosts on one interface. func (m *Monitor) lossLoop(ctx context.Context, st *InterfaceStatus, hosts []string) { m.logf("Starting packet loss monitoring for %s with %d hosts", st.Name, len(hosts)) time.Sleep(randomOffset()) tk := time.NewTicker(m.PacketLossPeriod) defer tk.Stop() for { select { case <-ctx.Done(): m.logf("Stopping packet loss monitoring for %s", st.Name) return case <-tk.C: m.lossTick(ctx, st, hosts) } } } // lossTick measures loss for every host concurrently and applies results. func (m *Monitor) lossTick(ctx context.Context, st *InterfaceStatus, hosts []string) { res := probeAll(hosts, func(host string) float64 { return m.lossProbe(ctx, st, host) }) m.lossApply(st, res) } // lossProbe measures loss for a host and updates per-host counters. func (m *Monitor) lossProbe( ctx context.Context, st *InterfaceStatus, host string, ) float64 { lp := m.lossPercent(ctx, st.Name, host) st.mu.Lock() defer st.mu.Unlock() if lp == 0 { st.Spin() return lp } lostPackets := int(math.Ceil(float64(m.PacketLossPings) * lp)) st.DroppedCount += lostPackets if lostPackets > 0 { st.LastDrop = time.Now() } st.LostPackets[host] += lostPackets m.notifyUI() return lp } // lossApply stores the round's loss results. func (m *Monitor) lossApply(st *InterfaceStatus, res map[string]float64) { st.mu.Lock() changed := lossChanged(st.Loss, res) for k, v := range res { st.Loss[k] = v } st.mu.Unlock() if changed { m.notifyUI() } } // lossChanged reports whether any host's loss moved beyond the threshold. func lossChanged(old, cur map[string]float64) bool { for host, newLoss := range cur { if oldLoss, ok := old[host]; !ok || math.Abs(oldLoss-newLoss) > lossChangeThreshold { return true } } return false } // tcpLoop monitors TCP connectivity for hosts on one interface. func (m *Monitor) tcpLoop(ctx context.Context, st *InterfaceStatus, hosts []string) { m.logf("Starting TCP monitoring for %s with %d hosts", st.Name, len(hosts)) time.Sleep(randomOffset()) tk := time.NewTicker(time.Second) defer tk.Stop() for { select { case <-ctx.Done(): m.logf("Stopping TCP monitoring for %s", st.Name) return case <-tk.C: m.tcpTick(st, hosts) } } } // tcpTick measures TCP latency for each host and redraws on change. func (m *Monitor) tcpTick(st *InterfaceStatus, hosts []string) { changed := false for _, hp := range hosts { if m.tcpProbe(st, hp) { changed = true } } if changed { m.notifyUI() } } // tcpProbe measures one host's latency, records it, and reports whether the // latency changed significantly. func (m *Monitor) tcpProbe(st *InterfaceStatus, hp string) bool { ms := float64(m.tcpDuration(st.Name, hp).Milliseconds()) st.mu.Lock() defer st.mu.Unlock() hist := st.TCP[hp] changed := tcpSignificant(hist, ms) if ms < float64(m.TCPTimeout.Milliseconds()) { st.Spin() } else { m.notifyUI() } if len(hist) >= m.StatsHistory { hist = hist[1:] } st.TCP[hp] = append(hist, ms) host := strings.Split(hp, ":")[0] if ms >= float64(m.TCPTimeout.Milliseconds()) { st.LostPackets[host]++ } return changed } // tcpSignificant reports whether ms differs meaningfully from the last // sample (or there is no prior sample). func tcpSignificant(hist []float64, ms float64) bool { if len(hist) == 0 { return true } return math.Abs(hist[len(hist)-1]-ms) > tcpChangeThreshold }