housecarl/internal/stats/collector_test.go

package stats

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"testing"
	"time"

	"housecarl/internal/api"
	"housecarl/internal/nft"
	"housecarl/internal/render"
)

// fakeHost is a /sys/class/net and a /proc to read from.
type fakeHost struct {
	t          *testing.T
	sys, proc  string
	table      *nft.Table
	tableError error
}

func newFakeHost(t *testing.T) *fakeHost {
	dir := t.TempDir()
	h := &fakeHost{t: t, sys: filepath.Join(dir, "sys"), proc: filepath.Join(dir, "proc"),
		table: &nft.Table{Counters: map[string]nft.Counter{}, QuotaUsage: nft.QuotaUsage{Quotas: map[string]int64{}, QuotaSets: map[string][]render.Element{}},
			DeviceSets: map[string][]render.Element{}}}
	h.write("proc/sys/net/netfilter/nf_conntrack_count", "120")
	h.write("proc/sys/net/netfilter/nf_conntrack_max", "65536")
	h.write("proc/self/stat", "4242 (housecarld) S 1 4242 4242 0 -1 4194560 3000 0 0 0 150 50 0 0 20 0 9 0 100 1000000 3000")
	h.write("proc/self/status", "Name:\thousecarld\nVmRSS:\t   20480 kB\nThreads:\t9\n")
	h.write("proc/meminfo", "MemTotal:        8000000 kB\nMemFree:         1000000 kB\nMemAvailable:    6000000 kB\n")
	h.write("proc/loadavg", "0.50 0.40 0.30 1/200 4242\n")
	return h
}

func (h *fakeHost) write(rel, content string) {
	p := filepath.Join(filepath.Dir(h.sys), rel)
	if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
		h.t.Fatal(err)
	}
	if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
		h.t.Fatal(err)
	}
}

func (h *fakeHost) iface(name string, rx, tx int64) {
	h.write("sys/"+name+"/statistics/rx_bytes", fmt.Sprint(rx))
	h.write("sys/"+name+"/statistics/tx_bytes", fmt.Sprint(tx))
	h.write("sys/"+name+"/statistics/rx_packets", fmt.Sprint(rx/100))
	h.write("sys/"+name+"/statistics/tx_packets", fmt.Sprint(tx/100))
}

func (h *fakeHost) sources() Sources {
	return Sources{
		SysNet: h.sys, Proc: h.proc,
		Table: func(context.Context) (*nft.Table, error) {
			if h.tableError != nil {
				return nil, h.tableError
			}
			return h.table, nil
		},
		Zones: func() map[string]string { return map[string]string{"eth0": "wan", "eth1": "lan"} },
		Classify: func(mark string) (string, bool) {
			switch mark {
			case "r1":
				return "forward", true
			case "p-input":
				return "input", true
			}
			return "input", false
		},
	}
}

func TestCollector(t *testing.T) {
	ctx := context.Background()
	h := newFakeHost(t)
	h.iface("eth0", 1000, 500)
	h.iface("eth1", 2000, 2000)
	h.iface("lo", 9999, 9999)
	h.table.Counters["r1"] = nft.Counter{Packets: 10, Bytes: 1000}
	h.table.Counters["r2"] = nft.Counter{Packets: 100, Bytes: 9000}
	h.table.Quotas["quota_r5"] = 1 << 20
	if h.table.QuotaSets == nil {
		h.table.QuotaSets = map[string][]render.Element{}
	}
	h.table.QuotaSets["quota_r6_v4"] = []render.Element{{Addr: "10.0.10.5", Used: 1000}}
	h.table.DeviceSets["stats_up_v4"] = []render.Element{{Addr: "10.0.10.5", Bytes: 5000}}

	store, err := OpenStore(":memory:")
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	loc := time.FixedZone("test", 0)
	now := time.Date(2026, 10, 19, 10, 0, 30, 0, loc)
	c := NewCollector(store, Options{TopN: 10, RetainMinute: 7 * 24 * time.Hour, RetainHour: 90 * 24 * time.Hour, RetainDay: 730 * 24 * time.Hour, Location: loc},
		h.sources(), slog.New(slog.NewTextHandler(io.Discard, nil)))
	c.SetClock(func() time.Time { return now })
	c.Start(ctx)

	// A minute of traffic.
	h.iface("eth0", 5000, 1500)
	h.table.Counters["r1"] = nft.Counter{Packets: 25, Bytes: 2500}
	h.table.Counters["p-input"] = nft.Counter{Packets: 3, Bytes: 180} // a new counter counts whole
	h.table.Quotas["quota_r5"] = 3 << 20
	h.table.QuotaSets["quota_r6_v4"] = []render.Element{{Addr: "10.0.10.5", Used: 2000}, {Addr: "10.0.10.6", Used: 500}} // a new entry counts whole
	h.table.DeviceSets["stats_up_v4"] = []render.Element{{Addr: "10.0.10.5", Bytes: 8000}}
	h.table.DeviceSets["stats_down_v6"] = []render.Element{{Addr: "2001:db8:1:2::", Bytes: 70000}}
	for range 2 {
		c.Packet(api.LogEntry{Time: now, Source: "198.51.100.7", Protocol: "tcp", DestPort: 22, Verdict: "drop"})
	}
	c.Packet(api.LogEntry{Time: now, Source: "198.51.100.8", Protocol: "icmp", ICMPType: "echo-request", Verdict: "drop"})
	c.Packet(api.LogEntry{Time: now, Source: "10.0.10.5", Protocol: "udp", DestPort: 53, Verdict: "accept"}) // not a drop
	now = time.Date(2026, 10, 19, 10, 1, 2, 0, loc)
	c.Sample(ctx)

	q := NewQuerier(c)
	sum := func(prefix string) map[string]int64 {
		es, err := q.Sums(ctx, prefix, time.Hour)
		if err != nil {
			t.Fatal(err)
		}
		m := map[string]int64{}
		for _, e := range es {
			m[e.Key] = e.Count
		}
		return m
	}
	check := func(what string, got, want int64) {
		t.Helper()
		if got != want {
			t.Errorf("%s = %d, want %d", what, got, want)
		}
	}
	check("rx:eth0", sum("rx:")["eth0"], 4000)
	check("rx:eth1 (no change)", sum("rx:")["eth1"], 0)
	check("rx:lo", sum("rx:")["lo"], 0)
	check("zrx:wan", sum("zrx:")["wan"], 4000)
	check("ztx:wan", sum("ztx:")["wan"], 1000)
	check("hits:r1", sum("hits:")["r1"], 15)
	check("hits:p-input", sum("hits:")["p-input"], 3)
	check("drop:forward", sum("drop:")["forward"], 15)
	check("drop:input", sum("drop:")["input"], 3)
	check("quota:r5", sum("quota:")["r5"], 2<<20)
	check("quota:r6 (per address: the sum)", sum("quota:")["r6"], 1500)

	s, err := q.Summary(ctx, time.Hour, 10)
	if err != nil {
		t.Fatal(err)
	}
	check("drops", s.Drops, 18)
	if len(s.Zones) == 0 || s.Zones[0].Name != "wan" || s.Zones[0].Received != 4000 {
		t.Errorf("zones %+v", s.Zones)
	}
	if len(s.BlockedSources) != 2 || s.BlockedSources[0] != (Entry{Key: "198.51.100.7", Count: 2}) {
		t.Errorf("blocked sources %+v", s.BlockedSources)
	}
	if len(s.BlockedPorts) != 2 || s.BlockedPorts[0].Key != "tcp/22" || s.BlockedPorts[1].Key != "icmp/echo-request" {
		t.Errorf("blocked ports %+v", s.BlockedPorts)
	}
	if len(s.DevicesUp) != 1 || s.DevicesUp[0] != (Entry{Key: "10.0.10.5", Count: 3000}) {
		t.Errorf("devices up %+v", s.DevicesUp)
	}
	if len(s.DevicesDown) != 1 || s.DevicesDown[0] != (Entry{Key: "2001:db8:1:2::/64", Count: 70000}) {
		t.Errorf("devices down %+v", s.DevicesDown)
	}
	check("peak connections", s.PeakConnections, 120)

	// Gauges average; the CPU needs two samples.
	tl, err := q.Timeline(ctx, time.Hour, "1m", []string{"g:" + GaugeRSS, "g:" + GaugeMem, "g:" + GaugeLoad, "rx:*"})
	if err != nil {
		t.Fatal(err)
	}
	last := func(i int) int64 {
		for j := len(tl.Times) - 1; j >= 0; j-- {
			if tl.Times[j] == time.Date(2026, 10, 19, 10, 0, 0, 0, loc).Unix() {
				return tl.Series[i].Values[j]
			}
		}
		return -2
	}
	check("rss", last(0), 20480*1024)
	check("mem", last(1), 2000000*1024)
	check("load", last(2), 50)
	if len(tl.Series) != 4 || tl.Series[3].Key != "rx:eth0" {
		t.Errorf("rx:* expanded to %+v", tl.Series[3:])
	}

	// Totals that go down started again: an interface came back, a quota
	// was reset. A table that can't be read is skipped, and its keys
	// aren't new the next time.
	h.iface("eth0", 100, 1500)
	h.table.Quotas["quota_r5"] = 1000
	h.tableError = fmt.Errorf("nft failed")
	now = time.Date(2026, 10, 19, 10, 2, 2, 0, loc)
	c.Sample(ctx)
	if st := c.Status(); st.SampleError == "" {
		t.Error("no sample error")
	}
	h.tableError = nil
	now = time.Date(2026, 10, 19, 10, 3, 2, 0, loc)
	c.Sample(ctx)
	check("rx:eth0 after a reset", sum("rx:")["eth0"], 4100)
	check("quota:r5 after a reset", sum("quota:")["r5"], 2<<20+1000)
	check("hits:r1 after a failed read", sum("hits:")["r1"], 15)

	// The hour ends: its top lists go to its day.
	now = time.Date(2026, 10, 19, 11, 0, 2, 0, loc)
	c.Sample(ctx)
	day, err := store.TopsAt(ctx, ResDay, time.Date(2026, 10, 19, 0, 0, 0, 0, loc).Unix(), KindBlockedSource)
	if err != nil || len(day) != 2 || day[0].Count != 2 {
		t.Errorf("the day's blocked sources: %+v %v", day, err)
	}
	if es, _ := q.Top(ctx, KindBlockedSource, 30*24*time.Hour, 5); len(es) != 2 {
		t.Errorf("30 days of blocked sources: %+v", es)
	}

	// A restart continues the hour's top lists.
	c.Packet(api.LogEntry{Time: now, Source: "203.0.113.9", Protocol: "tcp", DestPort: 23, Verdict: "drop"})
	c.write(ctx, &Batch{})
	c2 := NewCollector(store, c.opts, h.sources(), c.log)
	c2.SetClock(func() time.Time { return now })
	c2.Start(ctx)
	c2.Packet(api.LogEntry{Time: now, Source: "203.0.113.9", Protocol: "tcp", DestPort: 23, Verdict: "drop"})
	c2.write(ctx, &Batch{})
	hour, _ := store.TopsAt(ctx, ResHour, time.Date(2026, 10, 19, 11, 0, 0, 0, loc).Unix(), KindBlockedSource)
	if len(hour) != 1 || hour[0].Count != 2 {
		t.Errorf("after a restart, the hour's blocked sources: %+v", hour)
	}

	// The heatmap of drops.
	hm, err := q.Heatmap(ctx, "", 7*24*time.Hour)
	if err != nil {
		t.Fatal(err)
	}
	if hm.Avg[0][10] != 18 { // a Monday, 10:00
		t.Errorf("heatmap Monday 10:00 = %v", hm.Avg[0][10])
	}
}