nixos-router/pkgs/router-webui/collect.go

package main

import (
	"bufio"
	"context"
	"encoding/binary"
	"encoding/csv"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"math/rand/v2"
	"net"
	"net/netip"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

const sampleEvery = 5 * time.Second

// Collector samples the box every few seconds into the Store and keeps the
// latest snapshot for the dashboard.
type Collector struct {
	sys        System
	store      *Store
	storePath  string
	leasesPath string

	mu   sync.Mutex
	snap Snapshot

	prevIf   map[string][2]uint64
	prevCPU  [2]uint64 // busy, total
	prevTime time.Time
}

// Snapshot is the state right now.
type Snapshot struct {
	Time   time.Time
	Uplink UplinkStatus
	System SystemStatus
	Pings  []PingResult
}

type UplinkStatus struct {
	Device    string // pppoe-wan or the WAN interface
	Type      string // dhcp or pppoe
	Exists    bool
	Connected bool // an IPv4 address and the default route on the uplink
	Reachable bool // at least one public ping target answered
	Checked   bool // public ping targets are configured (Reachable means something)
	IPv4      []string
	IPv6      []string
	Gateway   string
	Since     time.Time
	Events    []Event
}

type SystemStatus struct {
	Uptime      time.Duration
	Load        string
	CPU         float64 // percent; NaN until the second sample
	MemUsed     uint64
	MemTotal    uint64
	Temperature float64 // °C, NaN if no sensor
	DiskUsed    uint64
	DiskTotal   uint64
}

type PingResult struct {
	Target  string // as configured ("gateway" or an address)
	Address string // what was pinged
	RTT     time.Duration
	OK      bool
}

func newCollector(sys System, leasesPath string) *Collector {
	path := filepath.Join(sys.StateDir, "webui", "metrics.gob")
	store, err := loadStore(path)
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("metric history: %v (starting empty)", err)
	}
	return &Collector{sys: sys, store: store, storePath: path, leasesPath: leasesPath}
}

func (c *Collector) Snapshot() Snapshot {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.snap
}

func (c *Collector) Save() {
	if err := os.MkdirAll(filepath.Dir(c.storePath), 0o700); err != nil {
		log.Printf("saving metric history: %v", err)
		return
	}
	if err := c.store.Save(c.storePath); err != nil {
		log.Printf("saving metric history: %v", err)
	}
}

// Run samples until ctx ends, saving the history every 15 minutes and at the end.
func (c *Collector) Run(ctx context.Context) {
	tick := time.NewTicker(sampleEvery)
	defer tick.Stop()
	save := time.NewTicker(15 * time.Minute)
	defer save.Stop()
	c.sample(time.Now())
	for {
		select {
		case <-ctx.Done():
			c.Save()
			return
		case <-save.C:
			c.Save()
		case now := <-tick.C:
			c.sample(now)
		}
	}
}

func (c *Collector) sample(now time.Time) {
	s, _ := loadSettings(c.sys.settingsPath())
	f := s.form()
	prefs := c.sys.loadPrefs()

	// Pings run while the rest is sampled.
	var pings []PingResult
	done := make(chan struct{})
	go func() {
		pings = pingAll(prefs.PingTargets, 2*time.Second)
		close(done)
	}()

	dt := now.Sub(c.prevTime).Seconds()
	counters := interfaceCounters()
	if c.prevIf != nil && dt > 0 {
		for dev, cur := range counters {
			prev, ok := c.prevIf[dev]
			if !ok || cur[0] < prev[0] || cur[1] < prev[1] {
				continue // new device or counters reset (a PPP reconnect)
			}
			c.store.Add("if:"+dev+":rx", now, float64(cur[0]-prev[0])/dt)
			c.store.Add("if:"+dev+":tx", now, float64(cur[1]-prev[1])/dt)
		}
	}
	c.prevIf, c.prevTime = counters, now

	sys := systemStatus()
	busy, total := cpuTimes()
	sys.CPU = math.NaN()
	if c.prevCPU[1] != 0 && total > c.prevCPU[1] {
		sys.CPU = 100 * float64(busy-c.prevCPU[0]) / float64(total-c.prevCPU[1])
	}
	c.prevCPU = [2]uint64{busy, total}
	c.store.Add("sys:cpu", now, sys.CPU)
	if sys.MemTotal > 0 {
		c.store.Add("sys:mem", now, 100*float64(sys.MemUsed)/float64(sys.MemTotal))
	}
	c.store.Add("sys:temp", now, sys.Temperature)

	up := uplinkStatus(f.WAN)
	<-done
	for _, p := range pings {
		loss := 100.0
		if p.OK {
			loss = 0
			c.store.Add("ping:"+p.Target+":rtt", now, float64(p.RTT.Microseconds())/1000)
		}
		c.store.Add("ping:"+p.Target+":loss", now, loss)
		if p.Target != "gateway" {
			up.Checked = true
			up.Reachable = up.Reachable || p.OK
		}
	}
	detail := strings.Join(up.IPv4, ", ")
	if !up.Connected {
		detail = ""
	}
	c.store.SetConnected(up.Connected, now, detail)
	_, up.Since, up.Events = c.store.Uplink()

	c.mu.Lock()
	c.snap = Snapshot{Time: now, Uplink: up, System: sys, Pings: pings}
	c.mu.Unlock()
}

// interfaceCounters reads rx/tx bytes of every interface but lo.
func interfaceCounters() map[string][2]uint64 {
	out := map[string][2]uint64{}
	entries, _ := os.ReadDir("/sys/class/net")
	for _, e := range entries {
		dev := e.Name()
		if dev == "lo" {
			continue
		}
		rx, err1 := readUint(filepath.Join("/sys/class/net", dev, "statistics/rx_bytes"))
		tx, err2 := readUint(filepath.Join("/sys/class/net", dev, "statistics/tx_bytes"))
		if err1 == nil && err2 == nil {
			out[dev] = [2]uint64{rx, tx}
		}
	}
	return out
}

func readUint(path string) (uint64, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return 0, err
	}
	return strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64)
}

func cpuTimes() (busy, total uint64) {
	f, err := os.Open("/proc/stat")
	if err != nil {
		return 0, 0
	}
	defer f.Close()
	line, _ := bufio.NewReader(f).ReadString('\n')
	fields := strings.Fields(line)
	if len(fields) < 5 || fields[0] != "cpu" {
		return 0, 0
	}
	for i, v := range fields[1:] {
		n, _ := strconv.ParseUint(v, 10, 64)
		total += n
		if i != 3 && i != 4 { // idle, iowait
			busy += n
		}
	}
	return busy, total
}

func systemStatus() SystemStatus {
	s := SystemStatus{Temperature: temperature()}
	if data, err := os.ReadFile("/proc/uptime"); err == nil {
		if f := strings.Fields(string(data)); len(f) > 0 {
			secs, _ := strconv.ParseFloat(f[0], 64)
			s.Uptime = time.Duration(secs) * time.Second
		}
	}
	if data, err := os.ReadFile("/proc/loadavg"); err == nil {
		if f := strings.Fields(string(data)); len(f) >= 3 {
			s.Load = strings.Join(f[:3], " ")
		}
	}
	if f, err := os.Open("/proc/meminfo"); err == nil {
		mem := map[string]uint64{}
		sc := bufio.NewScanner(f)
		for sc.Scan() {
			k, v, ok := strings.Cut(sc.Text(), ":")
			if ok {
				n, _ := strconv.ParseUint(strings.TrimSuffix(strings.TrimSpace(v), " kB"), 10, 64)
				mem[k] = n * 1024
			}
		}
		f.Close()
		s.MemTotal = mem["MemTotal"]
		if mem["MemAvailable"] <= s.MemTotal {
			s.MemUsed = s.MemTotal - mem["MemAvailable"]
		}
	}
	var st syscall.Statfs_t
	if syscall.Statfs("/", &st) == nil {
		s.DiskTotal = st.Blocks * uint64(st.Bsize)
		s.DiskUsed = (st.Blocks - st.Bfree) * uint64(st.Bsize)
	}
	return s
}

// temperature is the hottest thermal zone, or hwmon sensor if there is none.
func temperature() float64 {
	hottest := math.NaN()
	for _, pattern := range []string{"/sys/class/thermal/thermal_zone*/temp", "/sys/class/hwmon/hwmon*/temp*_input"} {
		files, _ := filepath.Glob(pattern)
		for _, f := range files {
			if v, err := readUint(f); err == nil && v > 0 && v < 150000 {
				if t := float64(v) / 1000; math.IsNaN(hottest) || t > hottest {
					hottest = t
				}
			}
		}
		if !math.IsNaN(hottest) {
			break
		}
	}
	return hottest
}

// uplinkDevice is where the router's IPv4 internet traffic leaves.
func uplinkDevice(wan WANForm) string {
	if wan.Type == "pppoe" {
		return "pppoe-wan"
	}
	return wan.Interface
}

func uplinkStatus(wan WANForm) UplinkStatus {
	u := UplinkStatus{Device: uplinkDevice(wan), Type: wan.Type}
	iface, err := net.InterfaceByName(u.Device)
	if err != nil {
		return u
	}
	u.Exists = true
	addrs, _ := iface.Addrs()
	for _, a := range addrs {
		ipnet, ok := a.(*net.IPNet)
		if !ok {
			continue
		}
		if ip4 := ipnet.IP.To4(); ip4 != nil {
			u.IPv4 = append(u.IPv4, ip4.String())
		} else if ipnet.IP.IsGlobalUnicast() {
			u.IPv6 = append(u.IPv6, ipnet.IP.String())
		}
	}
	gw, dev := defaultRoute()
	u.Connected = len(u.IPv4) > 0 && dev == u.Device
	if u.Connected {
		u.Gateway = gw
	}
	return u
}

// defaultRoute returns the IPv4 default route's gateway (for a point-to-point
// uplink: the peer) and device.
func defaultRoute() (gateway, dev string) {
	out, err := command("ip", "-j", "-4", "route", "show", "default")
	if err != nil {
		return "", ""
	}
	var routes []struct {
		Gateway string `json:"gateway"`
		Dev     string `json:"dev"`
	}
	if json.Unmarshal([]byte(out), &routes) != nil || len(routes) == 0 {
		return "", ""
	}
	gateway, dev = routes[0].Gateway, routes[0].Dev
	if gateway == "" && dev != "" {
		out, err := command("ip", "-j", "-4", "addr", "show", "dev", dev)
		if err != nil {
			return "", dev
		}
		var links []struct {
			AddrInfo []struct {
				Address string `json:"address"` // the peer, on point-to-point links
			} `json:"addr_info"`
		}
		if json.Unmarshal([]byte(out), &links) == nil && len(links) > 0 && len(links[0].AddrInfo) > 0 {
			gateway = links[0].AddrInfo[0].Address
		}
	}
	return gateway, dev
}

// pingAll pings every target once, in parallel. "gateway" is the default
// route's gateway (the ISP's end of the line).
func pingAll(targets []string, timeout time.Duration) []PingResult {
	out := make([]PingResult, len(targets))
	var wg sync.WaitGroup
	for i, t := range targets {
		out[i] = PingResult{Target: t, Address: t}
		if t == "gateway" {
			gw, _ := defaultRoute()
			out[i].Address = gw
		}
		addr, err := netip.ParseAddr(out[i].Address)
		if err != nil {
			continue
		}
		wg.Go(func() {
			rtt, err := ping(addr, timeout)
			out[i].RTT, out[i].OK = rtt, err == nil
		})
	}
	wg.Wait()
	return out
}

// ping sends one ICMP echo request over a raw socket (the UI runs as root).
func ping(addr netip.Addr, timeout time.Duration) (time.Duration, error) {
	network, request, reply := "ip4:icmp", byte(8), byte(0)
	if addr.Is6() && !addr.Is4In6() {
		network, request, reply = "ip6:ipv6-icmp", 128, 129
	}
	conn, err := net.ListenPacket(network, "")
	if err != nil {
		return 0, err
	}
	defer conn.Close()
	id := uint16(rand.Uint32())
	msg := make([]byte, 16)
	msg[0] = request
	binary.BigEndian.PutUint16(msg[4:], id)
	binary.BigEndian.PutUint16(msg[6:], 1)
	if request == 8 {
		// The kernel fills in the ICMPv6 checksum, not ICMPv4's.
		binary.BigEndian.PutUint16(msg[2:], checksum(msg))
	}
	start := time.Now()
	conn.SetDeadline(start.Add(timeout))
	if _, err := conn.WriteTo(msg, &net.IPAddr{IP: addr.AsSlice()}); err != nil {
		return 0, err
	}
	buf := make([]byte, 1500)
	for {
		n, from, err := conn.ReadFrom(buf)
		if err != nil {
			return 0, err
		}
		// Raw sockets see every ICMP message; ours carries our ID.
		if n >= 8 && buf[0] == reply && binary.BigEndian.Uint16(buf[4:]) == id {
			if ip, ok := from.(*net.IPAddr); ok && ip.IP.Equal(addr.AsSlice()) {
				return time.Since(start), nil
			}
		}
	}
}

func checksum(b []byte) uint16 {
	var sum uint32
	for i := 0; i+1 < len(b); i += 2 {
		sum += uint32(b[i])<<8 | uint32(b[i+1])
	}
	if len(b)%2 == 1 {
		sum += uint32(b[len(b)-1]) << 8
	}
	for sum>>16 != 0 {
		sum = sum&0xffff + sum>>16
	}
	return ^uint16(sum)
}

// Lease is an active DHCPv4 lease.
type Lease struct {
	Address  string
	MAC      string
	Hostname string
	Expires  time.Time
	LAN      string
}

// readLeases reads Kea's memfile: a CSV log in which the last line for an
// address wins. The previous file of a lease file cleanup comes first.
func readLeases(path string, lans []LANForm, now time.Time) []Lease {
	latest := map[string]Lease{}
	for _, p := range []string{path + ".completed", path + ".2", path} {
		f, err := os.Open(p)
		if err != nil {
			continue
		}
		parseLeases(f, latest, now)
		f.Close()
	}
	var out []Lease
	for _, l := range latest {
		if l.Expires.After(now) {
			l.LAN = lanOf(l.Address, lans)
			out = append(out, l)
		}
	}
	slices.SortFunc(out, func(a, b Lease) int {
		x, _ := netip.ParseAddr(a.Address)
		y, _ := netip.ParseAddr(b.Address)
		return x.Compare(y)
	})
	return out
}

func parseLeases(r io.Reader, latest map[string]Lease, now time.Time) {
	cr := csv.NewReader(r)
	cr.FieldsPerRecord = -1
	header, err := cr.Read()
	if err != nil {
		return
	}
	col := map[string]int{}
	for i, h := range header {
		col[h] = i
	}
	get := func(rec []string, name string) string {
		if i, ok := col[name]; ok && i < len(rec) {
			return rec[i]
		}
		return ""
	}
	for {
		rec, err := cr.Read()
		if err == io.EOF {
			return
		}
		if err != nil {
			continue
		}
		if len(rec) > 0 && rec[0] == header[0] {
			continue // a header repeated by a cleanup
		}
		addr := get(rec, "address")
		expire, _ := strconv.ParseInt(get(rec, "expire"), 10, 64)
		lifetime, _ := strconv.ParseInt(get(rec, "valid_lifetime"), 10, 64)
		state := get(rec, "state")
		l := Lease{
			Address:  addr,
			MAC:      get(rec, "hwaddr"),
			Hostname: strings.ReplaceAll(get(rec, "hostname"), "&#x2c", ","),
			Expires:  time.Unix(expire, 0),
		}
		if lifetime == 0 || (state != "" && state != "0") {
			l.Expires = time.Time{} // released, declined or reclaimed
		}
		latest[addr] = l
	}
}

func lanOf(addr string, lans []LANForm) string {
	a, err := netip.ParseAddr(addr)
	if err != nil {
		return ""
	}
	for _, l := range lans {
		prefix := 24
		if p, err := strconv.Atoi(l.Prefix); err == nil {
			prefix = p
		}
		if pfx, err := netip.ParsePrefix(fmt.Sprintf("%s/%d", l.Address, prefix)); err == nil && pfx.Masked().Contains(a) {
			return l.Name
		}
	}
	return ""
}