rtnetmon/internal/monitor/loops.go

//go:build linux

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
}