jarl/internal/stats/collector_test.go

package stats

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"net/netip"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"
)

// fakeNamed serves a statistics channel whose counters the test sets.
type fakeNamed struct {
	mu      sync.Mutex
	queries int64
	nx      int64
	srv     *httptest.Server
}

func newFakeNamed(t *testing.T) *fakeNamed {
	f := &fakeNamed{}
	f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		fmt.Fprintf(w, `{"boot-time":"2026-10-05T08:00:00Z","opcodes":{"QUERY":%d},"rcodes":{"NOERROR":%d,"NXDOMAIN":%d},"qtypes":{"A":%d},"nsstats":{},"views":{}}`,
			f.queries, f.queries-f.nx, f.nx, f.queries)
	}))
	t.Cleanup(f.srv.Close)
	return f
}

func (f *fakeNamed) add(q, nx int64) {
	f.mu.Lock()
	f.queries += q
	f.nx += nx
	f.mu.Unlock()
}

type clock struct{ t time.Time }

func (c *clock) now() time.Time { return c.t }

func logLine(t time.Time, client, view, name, typ, rcode string) string {
	v := ""
	if view != "" {
		v = "view " + view + ": "
	}
	return fmt.Sprintf("%s client @0x7f0000000000 %s#5353 (%s): %sresponse: %s IN %s %s 1 0 1 +E(0)K (10.10.0.2)\n",
		t.Format(logTimeLayout), client, name, v, name, typ, rcode)
}

func TestCollector(t *testing.T) {
	ctx := context.Background()
	loc := time.FixedZone("test", 2*3600)
	dir := t.TempDir()
	logFile := filepath.Join(dir, "responses.log")
	store, err := OpenStore(filepath.Join(dir, "stats.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	named := newFakeNamed(t)
	clk := &clock{time.Date(2026, 10, 5, 10, 0, 30, 0, loc)} // a Monday
	zones := func() map[string][]string {
		return map[string][]string{"internal": {"adbraun.lol", "0.10.10.in-addr.arpa"}}
	}
	opts := Options{QueryLog: true, LogFile: logFile, StatisticsChannel: "unused", TopN: 3,
		RetainMinute: 7 * 24 * time.Hour, RetainHour: 90 * 24 * time.Hour, RetainDay: 730 * 24 * time.Hour,
		Exclude: []netip.Prefix{netip.MustParsePrefix("10.10.0.250/32")}, Location: loc}
	log := slog.New(slog.NewTextHandler(io.Discard, nil))
	newCollector := func() *Collector {
		c := NewCollector(store, opts, zones, log)
		c.poller.URL = named.srv.URL
		c.now = clk.now
		c.start(ctx)
		return c
	}
	write := func(lines ...string) {
		f, err := os.OpenFile(logFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
		if err != nil {
			t.Fatal(err)
		}
		for _, l := range lines {
			f.WriteString(l)
		}
		f.Close()
	}
	c := newCollector()

	// Minute 10:00: four responses (one from an excluded monitor).
	base := time.Date(2026, 10, 5, 10, 0, 40, 0, loc)
	write(logLine(base, "10.10.0.37", "internal", "nas.adbraun.lol", "A", "NOERROR"),
		logLine(base, "10.10.0.37", "internal", "nas.adbraun.lol", "AAAA", "NOERROR"),
		logLine(base, "10.10.0.42", "internal", "nope.adbraun.lol", "A", "NXDOMAIN"),
		logLine(base, "10.10.0.42", "internal", "debian.org", "A", "NOERROR"),
		logLine(base, "10.10.0.250", "internal", "nas.adbraun.lol", "A", "NOERROR"))
	c.readLog()
	named.add(5, 1)
	clk.t = time.Date(2026, 10, 5, 10, 1, 2, 0, loc)
	c.flush(ctx, false)

	m := time.Date(2026, 10, 5, 10, 0, 0, 0, loc).Unix()
	got, err := store.sums(ctx, ResMinute, m, m+60, "")
	if err != nil {
		t.Fatal(err)
	}
	want := map[string]int64{"queries": 5, "rcode:NOERROR": 4, "rcode:NXDOMAIN": 1, "qtype:A": 5, "responses": 4,
		"view:internal": 4, "zone:adbraun.lol": 3, "zone:-": 1}
	for k, v := range want {
		if got[k] != v {
			t.Errorf("minute %s = %d, want %d (all: %v)", k, got[k], v, got)
		}
	}
	names, _ := store.TopsAt(ctx, ResHour, m, KindName)
	if len(names) != 3 || names[0] != (Entry{Key: "nas.adbraun.lol", Count: 2}) {
		t.Errorf("hour names = %v", names)
	}
	if nx, _ := store.TopsAt(ctx, ResHour, m, KindNXDomain); len(nx) != 1 || nx[0].Key != "nope.adbraun.lol" {
		t.Errorf("hour nxdomain = %v", nx)
	}

	// Into the next hour: the 10:00 hour is finished and added to the day.
	write(logLine(time.Date(2026, 10, 5, 10, 59, 50, 0, loc), "10.10.0.37", "internal", "nas.adbraun.lol", "A", "NOERROR"),
		logLine(time.Date(2026, 10, 5, 11, 0, 1, 0, loc), "10.10.0.99", "internal", "printer.adbraun.lol", "A", "NOERROR"))
	c.readLog()
	named.add(2, 0)
	clk.t = time.Date(2026, 10, 5, 11, 0, 2, 0, loc)
	c.flush(ctx, false)
	day := time.Date(2026, 10, 5, 0, 0, 0, 0, loc).Unix()
	if d, _ := store.TopsAt(ctx, ResDay, day, KindName); len(d) == 0 || d[0] != (Entry{Key: "nas.adbraun.lol", Count: 3}) {
		t.Errorf("day names after the hour = %v", d)
	}
	if v, _ := store.sums(ctx, ResDay, day, day+1, "queries"); v["queries"] != 7 {
		t.Errorf("day queries = %v", v)
	}

	// Restart in the middle of hour 11. Lines from hour 10 that were
	// still unread are added to hour 10 and the day, not replacing them.
	c.flush(ctx, true)
	c2 := newCollector()
	write(logLine(time.Date(2026, 10, 5, 10, 59, 59, 0, loc), "10.10.0.37", "internal", "nas.adbraun.lol", "A", "NOERROR"),
		logLine(time.Date(2026, 10, 5, 11, 2, 0, 0, loc), "10.10.0.99", "internal", "printer.adbraun.lol", "A", "NOERROR"))
	c2.readLog()
	clk.t = time.Date(2026, 10, 5, 11, 3, 2, 0, loc)
	c2.flush(ctx, false)
	if h, _ := store.TopsAt(ctx, ResHour, m, KindName); len(h) == 0 || h[0] != (Entry{Key: "nas.adbraun.lol", Count: 4}) {
		t.Errorf("hour 10 after late line = %v", h)
	}
	if d, _ := store.TopsAt(ctx, ResDay, day, KindName); d[0] != (Entry{Key: "nas.adbraun.lol", Count: 4}) {
		t.Errorf("day after late line = %v", d)
	}
	h11 := time.Date(2026, 10, 5, 11, 0, 0, 0, loc).Unix()
	if h, _ := store.TopsAt(ctx, ResHour, h11, KindName); len(h) != 1 || h[0] != (Entry{Key: "printer.adbraun.lol", Count: 2}) {
		t.Errorf("hour 11 continued after restart = %v", h)
	}

	// The read side.
	q := &Querier{Store: store, Opts: opts, Now: clk.now}
	s, err := q.Summary(ctx, 24*time.Hour, 5)
	if err != nil {
		t.Fatal(err)
	}
	if s.Queries != 7 || s.Resolution != ResMinute || s.Tops[KindName][0].Key != "nas.adbraun.lol" || s.Peak == nil || s.Peak.Queries != 5 {
		t.Errorf("summary = %+v", s)
	}
	if s.NXDomainRatio == nil || *s.NXDomainRatio != 1.0/7 {
		t.Errorf("nxdomain ratio = %v", s.NXDomainRatio)
	}
	tl, err := q.Timeline(ctx, 2*time.Hour, "auto", []string{"queries", "rcode:*"})
	if err != nil {
		t.Fatal(err)
	}
	if tl.Step != "1m" || len(tl.Series) != 3 {
		t.Fatalf("timeline step %s series %d", tl.Step, len(tl.Series))
	}
	var total int64
	for _, v := range tl.Series[0].Values {
		total += v
	}
	if tl.Series[0].Key != "queries" || total != 7 {
		t.Errorf("timeline queries = %v", tl.Series[0])
	}
	if tl, _ := q.Timeline(ctx, 30*24*time.Hour, "auto", []string{"queries"}); tl.Step != "6h" {
		t.Errorf("30d step = %s", tl.Step)
	}
	hm, err := q.Heatmap(ctx, "", 7*24*time.Hour)
	if err != nil {
		t.Fatal(err)
	}
	// The counters polled at 11:00:02 belong to the minute before.
	if hm.Avg[0][10] != 7 || hm.Avg[0][11] != 0 || hm.Samples[0][11] != 1 || hm.Avg[3][3] != -1 {
		t.Errorf("heatmap Monday 10h=%v 11h=%v Thu 3h=%v", hm.Avg[0][10], hm.Avg[0][11], hm.Avg[3][3])
	}
	b, _ := q.Breakdown(ctx, "zone", 24*time.Hour)
	if len(b) != 2 || b[0] != (Entry{Key: "adbraun.lol", Count: 7}) || b[1] != (Entry{Key: OtherZone, Count: 1}) {
		t.Errorf("zone breakdown = %v", b)
	}

	// Retention: minute data older than the retention disappears.
	opts.RetainMinute = time.Hour
	c3 := NewCollector(store, opts, zones, log)
	c3.now = func() time.Time { return clk.t.Add(2 * time.Hour) }
	c3.prune(ctx)
	if v, _ := store.sums(ctx, ResMinute, 0, 1<<40, "queries"); v["queries"] != 0 {
		t.Errorf("minute data survived pruning: %v", v)
	}
	if v, _ := store.sums(ctx, ResHour, 0, 1<<40, "queries"); v["queries"] != 7 {
		t.Errorf("hour data was pruned: %v", v)
	}
}

func TestAnonymizeAndParseSince(t *testing.T) {
	if k := clientKey(netip.MustParseAddr("10.10.0.37"), true); k != "10.10.0.0/24" {
		t.Errorf("v4 = %s", k)
	}
	if k := clientKey(netip.MustParseAddr("2001:db8:1:2::7"), true); k != "2001:db8:1::/48" {
		t.Errorf("v6 = %s", k)
	}
	for in, want := range map[string]time.Duration{"": 24 * time.Hour, "90m": 90 * time.Minute, "7d": 7 * 24 * time.Hour, "2w": 14 * 24 * time.Hour} {
		if d, err := ParseSince(in); err != nil || d != want {
			t.Errorf("ParseSince(%q) = %v, %v", in, d, err)
		}
	}
	for _, bad := range []string{"x", "-1h", "0d", "1.5d"} {
		if _, err := ParseSince(bad); err == nil {
			t.Errorf("ParseSince(%q) accepted", bad)
		}
	}
}