Statistics: traffic, drops, devices and connections over time

internal/stats samples every minute what the kernel already counts: the
interfaces' counters (per zone too), Housecarl's table (hits per rule,
built-in and policy; drops by chain; quotas; per-device sets), the
connection tracking table, and housecarld's and the host's memory, CPU
and load. Logged drops feed top lists of sources and what they tried.
Aggregates go to a database of their own (stats.db; per minute 7 days,
per hour 90 days, per day 2 years), with the store and queries from
Jarl's.

Traffic per inside device is a setting (stats_per_device): counting sets
stats_up/down_v4/v6 and rules at the top of the forward chain, IPv6 per
/64, carried across reloads. Set names starting with meter_, conn_,
quota_ or stats_ are reserved (set_name_reserved).

GET /stats/{status,summary,timeline,top,hits,heatmap}, hc stats, a
Statistics page with Jarl's charts, two charts on the dashboard, and a
conntrack_full warning. e2e: 188 checks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/stats/collector.go

@@ -0,0 +1,546 @@
+package stats
+
+import (
+	"cmp"
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"housecarl/internal/api"
+	"housecarl/internal/nft"
+)
+
+// Resolutions and retention, and the sources the figures come from.
+type Options struct {
+	TopN         int
+	RetainMinute time.Duration
+	RetainHour   time.Duration
+	RetainDay    time.Duration
+	Location     *time.Location // for days (default: local time)
+	// Interval is how often the collector samples (default: a minute).
+	Interval time.Duration
+}
+
+// Sources are what the collector reads. Nil functions are left out.
+type Sources struct {
+	// Table reads Housecarl's table: the counters by mark (totals across
+	// reloads), the quotas, the per-device sets.
+	Table func(context.Context) (*nft.Table, error)
+	// Zones maps the interfaces to their zones.
+	Zones func() map[string]string
+	// Classify says which chain a mark's counter is in and whether what
+	// it counts is dropped ("r12" → forward, true).
+	Classify func(mark string) (chain string, drop bool)
+	// SysNet and Proc are /sys/class/net and /proc (tests point them
+	// elsewhere).
+	SysNet, Proc string
+}
+
+// Counter keys. Traffic: rx:IF, tx:IF (bytes), rxp:IF, txp:IF (packets),
+// zrx:ZONE, ztx:ZONE (bytes received from and sent to a zone's
+// interfaces). Hits: hits:MARK (packets), hitb:MARK (bytes). Drops:
+// drop:CHAIN and drops (packets). Quotas: quota:r12 (bytes). Gauges
+// (g:KEY, see gauge) are below.
+const (
+	GaugeConntrack    = "conntrack"
+	GaugeConntrackMax = "conntrack_max"
+	GaugeRSS          = "rss"  // housecarld's resident memory, bytes
+	GaugeCPU          = "cpu"  // housecarld's CPU, tenths of a percent of one CPU
+	GaugeMem          = "mem"  // the host's memory in use, bytes
+	GaugeLoad         = "load" // the host's 1-minute load average, times 100
+)
+
+// Top-list kinds: per hour and per day, the TopN largest.
+const (
+	KindBlockedSource = "blocked_src"  // logged drops by source address
+	KindBlockedPort   = "blocked_port" // logged drops by what they tried: tcp/22, icmp
+	KindDeviceUp      = "device_up"    // bytes sent to the internet, by inside address
+	KindDeviceDown    = "device_down"  // bytes received from it
+	// A per-address quota's use by address is "quota_r12".
+)
+
+// Kinds are the top lists Top accepts (and quota_rN).
+var Kinds = []string{KindBlockedSource, KindBlockedPort, KindDeviceUp, KindDeviceDown}
+
+// Status describes the collector's health.
+type Status struct {
+	Enabled     bool       `json:"enabled"`
+	Database    string     `json:"database,omitempty"`
+	Size        int64      `json:"size"`
+	LastSample  *time.Time `json:"last_sample,omitempty"`
+	SampleError string     `json:"sample_error,omitempty"`
+	FirstData   *time.Time `json:"first_data,omitempty"`
+	// PerDevice: the firewall counts traffic per device (a setting).
+	PerDevice bool `json:"per_device"`
+	Retention struct {
+		Minute string `json:"minute"`
+		Hour   string `json:"hour"`
+		Day    string `json:"day"`
+	} `json:"retention"`
+}
+
+// Latest are the newest gauge values.
+type Latest struct {
+	At           time.Time `json:"at"`
+	Conntrack    int64     `json:"conntrack"`
+	ConntrackMax int64     `json:"conntrack_max"`
+	RSS          int64     `json:"rss"`
+	CPUPermille  int64     `json:"cpu_permille"`
+	MemUsed      int64     `json:"mem_used"`
+	Load         float64   `json:"load"`
+}
+
+// Collector gathers statistics into a Store.
+type Collector struct {
+	opts  Options
+	src   Sources
+	store *Store
+	log   *slog.Logger
+	now   func() time.Time
+
+	// Each source's last totals: the next sample stores the difference.
+	ifs, table deltas
+	cpuTicks   int64
+	cpuAt      time.Time
+	lastPrune  time.Time
+
+	mu     sync.Mutex
+	hours  map[int64]map[string]map[string]int64 // live hours: hour -> kind -> key -> count
+	status Status
+	latest *Latest
+}
+
+// maxKeys bounds each kind of an hour's top list in memory.
+const maxKeys = 100_000
+
+// NewCollector creates a collector. Run starts it.
+func NewCollector(store *Store, opts Options, src Sources, log *slog.Logger) *Collector {
+	if opts.Location == nil {
+		opts.Location = time.Local
+	}
+	if log == nil {
+		log = slog.Default()
+	}
+	if opts.TopN <= 0 {
+		opts.TopN = 100
+	}
+	if opts.Interval <= 0 {
+		opts.Interval = time.Minute
+	}
+	if src.SysNet == "" {
+		src.SysNet = "/sys/class/net"
+	}
+	if src.Proc == "" {
+		src.Proc = "/proc"
+	}
+	c := &Collector{opts: opts, src: src, store: store, log: log, now: time.Now, hours: map[int64]map[string]map[string]int64{}}
+	c.status.Enabled = true
+	c.status.Database = store.Path()
+	c.status.Retention.Minute = humanDuration(opts.RetainMinute)
+	c.status.Retention.Hour = humanDuration(opts.RetainHour)
+	c.status.Retention.Day = humanDuration(opts.RetainDay)
+	return c
+}
+
+// SetClock replaces the clock (tests).
+func (c *Collector) SetClock(now func() time.Time) { c.now = now }
+
+// Store returns the database.
+func (c *Collector) Store() *Store { return c.store }
+
+// Options returns the collector's options.
+func (c *Collector) Options() Options { return c.opts }
+
+// Status returns the collector's health.
+func (c *Collector) Status() Status {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	s := c.status
+	s.Size = c.store.Size()
+	return s
+}
+
+// Latest returns the newest gauges (nil before the first sample).
+func (c *Collector) Latest() *Latest {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.latest == nil {
+		return nil
+	}
+	l := *c.latest
+	return &l
+}
+
+func (c *Collector) minuteOf(t time.Time) int64 { return t.Unix() - t.Unix()%60 }
+func (c *Collector) hourOf(t time.Time) int64   { return t.Unix() - t.Unix()%3600 }
+
+// dayOf returns the start of the local day containing t.
+func (c *Collector) dayOf(t time.Time) int64 {
+	t = t.In(c.opts.Location)
+	return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, c.opts.Location).Unix()
+}
+
+// Run collects until ctx is done.
+func (c *Collector) Run(ctx context.Context) {
+	c.Start(ctx)
+	for {
+		t := time.NewTimer(time.Until(c.nextSample()))
+		select {
+		case <-ctx.Done():
+			t.Stop()
+			c.write(context.Background(), &Batch{})
+			return
+		case <-t.C:
+			c.Sample(ctx)
+		}
+	}
+}
+
+// nextSample: two seconds into the next minute, so the minute that just
+// ended is complete (or after the interval, when it isn't a minute).
+func (c *Collector) nextSample() time.Time {
+	if c.opts.Interval != time.Minute {
+		return c.now().Add(c.opts.Interval)
+	}
+	return time.Unix(c.minuteOf(c.now())+62, 0)
+}
+
+// Start takes the first sample (the baseline the next ones are counted
+// from), continues the current hour's top lists, and adds hours that
+// ended while housecarld wasn't running to their days.
+func (c *Collector) Start(ctx context.Context) {
+	if v, _ := c.store.Meta(ctx, "first_data"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil {
+			t := time.Unix(n, 0)
+			c.status.FirstData = &t
+		}
+	}
+	cur := c.hourOf(c.now())
+	b := &Batch{}
+	merged, _ := c.store.Meta(ctx, "merged_through")
+	if last, err := strconv.ParseInt(merged, 10, 64); err == nil {
+		for h := max(last+3600, cur-48*3600); h < cur; h += 3600 {
+			kinds, _ := c.store.KindsAt(ctx, ResHour, h)
+			for _, k := range kinds {
+				if es, err := c.store.TopsAt(ctx, ResHour, h, k); err == nil && len(es) > 0 {
+					b.AddTops(ResDay, c.dayOf(time.Unix(h, 0)), k, es, 3*c.opts.TopN)
+				}
+			}
+		}
+	}
+	b.SetMeta("merged_through", strconv.FormatInt(cur-3600, 10))
+	kinds, _ := c.store.KindsAt(ctx, ResHour, cur)
+	c.mu.Lock()
+	for _, k := range kinds {
+		if es, err := c.store.TopsAt(ctx, ResHour, cur, k); err == nil {
+			m := c.topsMap(cur, k)
+			for _, e := range es {
+				m[e.Key] = e.Count
+			}
+		}
+	}
+	c.mu.Unlock()
+	if err := c.store.Commit(ctx, b); err != nil {
+		c.log.Warn("statistics: catching up failed", "err", err)
+	}
+	c.sample(ctx, -1)
+	c.lastPrune = c.now()
+	c.prune(ctx)
+	c.log.Info("statistics collector started", "database", c.store.Path())
+}
+
+// Sample reads every source and writes what changed since the last
+// sample.
+func (c *Collector) Sample(ctx context.Context) {
+	now := c.now()
+	// The interval that just ended: with a minute's interval, sampling two
+	// seconds into a minute, the previous minute.
+	minute := c.minuteOf(now.Add(-c.opts.Interval / 2))
+	vals := c.sample(ctx, minute)
+	b := &Batch{}
+	if len(vals) > 0 {
+		t := time.Unix(minute, 0)
+		b.AddCounters(ResMinute, minute, vals)
+		b.AddCounters(ResHour, c.hourOf(t), vals)
+		b.AddCounters(ResDay, c.dayOf(t), vals)
+		c.mu.Lock()
+		if c.status.FirstData == nil {
+			c.status.FirstData = &t
+			b.SetMeta("first_data", strconv.FormatInt(minute, 10))
+		}
+		c.mu.Unlock()
+	}
+	c.write(ctx, b)
+	if now.Sub(c.lastPrune) >= time.Hour {
+		c.lastPrune = now
+		c.prune(ctx)
+	}
+}
+
+// write adds the live hours' top lists to b, ends the hours that are
+// over, and commits.
+func (c *Collector) write(ctx context.Context, b *Batch) {
+	c.mu.Lock()
+	cur := c.hourOf(c.now())
+	for h, kinds := range c.hours {
+		for k, m := range kinds {
+			b.SetTops(ResHour, h, k, top(m, c.opts.TopN))
+		}
+		if h < cur {
+			// The hour is over: add it to its day.
+			for k, m := range kinds {
+				b.AddTops(ResDay, c.dayOf(time.Unix(h, 0)), k, top(m, c.opts.TopN), 3*c.opts.TopN)
+			}
+			delete(c.hours, h)
+			b.SetMeta("merged_through", strconv.FormatInt(h, 10))
+		}
+	}
+	c.mu.Unlock()
+	if err := c.store.Commit(ctx, b); err != nil {
+		c.log.Error("statistics: writing failed", "err", err)
+	}
+}
+
+// deltas turn a source's totals into what changed since its last read.
+type deltas struct {
+	prev, next map[string]int64
+	based      bool
+}
+
+// of returns how much key grew. The first read is the baseline (nothing
+// counts); after it, a new key counts whole (a new interface, rule or
+// device starts from zero), and a total that went down started again (an
+// interface came back, a quota was reset, a device's entry expired).
+func (d *deltas) of(key string, cur int64) int64 {
+	if d.next == nil {
+		d.next = map[string]int64{}
+	}
+	d.next[key] = cur
+	if !d.based {
+		return 0
+	}
+	p, ok := d.prev[key]
+	if !ok || cur < p {
+		return cur
+	}
+	return cur - p
+}
+
+// done keeps this read's totals for the next (a failed read skips it, so
+// its keys aren't new next time).
+func (d *deltas) done() { d.prev, d.next, d.based = d.next, nil, true }
+
+// sample reads the sources; minute is the bucket the changes go to (-1:
+// only the baseline). It returns the counters.
+func (c *Collector) sample(ctx context.Context, minute int64) map[string]int64 {
+	now := c.now()
+	vals := map[string]int64{}
+	add := func(k string, v int64) {
+		if v != 0 {
+			vals[k] += v
+		}
+	}
+	gauge := func(k string, v int64) {
+		vals["g:"+k] += v
+		vals["gn:"+k]++
+	}
+	tops := map[string]map[string]int64{}
+	addTop := func(kind, key string, v int64) {
+		if v <= 0 {
+			return
+		}
+		if tops[kind] == nil {
+			tops[kind] = map[string]int64{}
+		}
+		tops[kind][key] += v
+	}
+	var errs []string
+
+	// Interfaces, and their zones.
+	var zones map[string]string
+	if c.src.Zones != nil {
+		zones = c.src.Zones()
+	}
+	for name, ic := range readInterfaces(c.src.SysNet) {
+		rx, tx := c.ifs.of("rx:"+name, ic.RxBytes), c.ifs.of("tx:"+name, ic.TxBytes)
+		add("rx:"+name, rx)
+		add("tx:"+name, tx)
+		add("rxp:"+name, c.ifs.of("rxp:"+name, ic.RxPackets))
+		add("txp:"+name, c.ifs.of("txp:"+name, ic.TxPackets))
+		if z := zones[name]; z != "" {
+			add("zrx:"+z, rx)
+			add("ztx:"+z, tx)
+		}
+	}
+	c.ifs.done()
+
+	// Housecarl's table: hits, drops, quotas, devices.
+	if c.src.Table != nil {
+		t, err := c.src.Table(ctx)
+		switch {
+		case err == nil:
+			for mark, ctr := range t.Counters {
+				p := c.table.of("hits:"+mark, int64(ctr.Packets))
+				add("hits:"+mark, p)
+				add("hitb:"+mark, c.table.of("hitb:"+mark, int64(ctr.Bytes)))
+				if c.src.Classify != nil && p > 0 {
+					if chain, drop := c.src.Classify(mark); drop {
+						add("drop:"+chain, p)
+						add("drops", p)
+					}
+				}
+			}
+			for name, used := range t.Quotas {
+				add("quota:"+strings.TrimPrefix(name, "quota_"), c.table.of("quota:"+name, used))
+			}
+			for set, es := range t.QuotaSets {
+				kind := strings.TrimSuffix(strings.TrimSuffix(set, "_v4"), "_v6")
+				for _, e := range es {
+					addTop(kind, e.Addr, c.table.of(set+"/"+e.Addr, e.Used))
+				}
+			}
+			for set, es := range t.DeviceSets {
+				kind := map[bool]string{true: KindDeviceUp, false: KindDeviceDown}[strings.HasPrefix(set, "stats_up")]
+				for _, e := range es {
+					key := e.Addr
+					if strings.HasSuffix(set, "_v6") {
+						key += "/64"
+					}
+					addTop(kind, key, c.table.of(set+"/"+e.Addr, int64(e.Bytes)))
+				}
+			}
+			c.table.done()
+		case errors.Is(err, nft.ErrNoTable):
+		default:
+			errs = append(errs, "reading the table: "+err.Error())
+		}
+	}
+
+	// Gauges.
+	l := &Latest{At: now}
+	if n, m, ok := readConntrack(c.src.Proc); ok {
+		l.Conntrack, l.ConntrackMax = n, m
+		gauge(GaugeConntrack, n)
+		gauge(GaugeConntrackMax, m)
+	}
+	if rss, ticks, ok := readSelf(c.src.Proc); ok {
+		l.RSS = rss
+		gauge(GaugeRSS, rss)
+		if !c.cpuAt.IsZero() && ticks >= c.cpuTicks {
+			secs := max(int64(now.Sub(c.cpuAt)/time.Second), 1)
+			l.CPUPermille = (ticks - c.cpuTicks) * 1000 / clockTicks / secs
+			gauge(GaugeCPU, l.CPUPermille)
+		}
+		c.cpuTicks, c.cpuAt = ticks, now
+	}
+	if mem, load, ok := readHost(c.src.Proc); ok {
+		l.MemUsed, l.Load = mem, load
+		gauge(GaugeMem, mem)
+		gauge(GaugeLoad, int64(load*100))
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.latest = l
+	c.status.LastSample = &now
+	c.status.SampleError = strings.Join(errs, "; ")
+	if minute < 0 {
+		return nil
+	}
+	h := c.hourOf(time.Unix(minute, 0))
+	for kind, m := range tops {
+		for k, v := range m {
+			c.addTop(h, kind, k, v)
+		}
+	}
+	return vals
+}
+
+// Packet counts a packet the firewall logged: drops by source and by
+// what they tried.
+func (c *Collector) Packet(e api.LogEntry) {
+	if e.Verdict != "drop" && e.Verdict != "reject" {
+		return
+	}
+	what := e.Protocol
+	switch {
+	case e.DestPort > 0:
+		what = fmt.Sprintf("%s/%d", e.Protocol, e.DestPort)
+	case e.ICMPType != "":
+		what = e.Protocol + "/" + e.ICMPType
+	}
+	t := e.Time
+	if t.IsZero() {
+		t = c.now()
+	}
+	h := c.hourOf(t)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if e.Source != "" {
+		c.addTop(h, KindBlockedSource, e.Source, 1)
+	}
+	if what != "" {
+		c.addTop(h, KindBlockedPort, what, 1)
+	}
+}
+
+// addTop counts into an hour's top list. Called with c.mu held.
+func (c *Collector) addTop(hour int64, kind, key string, n int64) {
+	m := c.topsMap(hour, kind)
+	if _, ok := m[key]; !ok && len(m) >= maxKeys {
+		return
+	}
+	m[key] += n
+}
+
+func (c *Collector) topsMap(hour int64, kind string) map[string]int64 {
+	h := c.hours[hour]
+	if h == nil {
+		h = map[string]map[string]int64{}
+		c.hours[hour] = h
+	}
+	m := h[kind]
+	if m == nil {
+		m = map[string]int64{}
+		h[kind] = m
+	}
+	return m
+}
+
+// top returns the n largest entries of m.
+func top(m map[string]int64, n int) []Entry {
+	out := make([]Entry, 0, len(m))
+	for k, v := range m {
+		out = append(out, Entry{Key: k, Count: v})
+	}
+	slices.SortFunc(out, func(a, b Entry) int {
+		if a.Count != b.Count {
+			return cmp.Compare(b.Count, a.Count)
+		}
+		return cmp.Compare(a.Key, b.Key)
+	})
+	if len(out) > n {
+		out = out[:n]
+	}
+	return out
+}
+
+func (c *Collector) prune(ctx context.Context) {
+	now := c.now().Unix()
+	for res, keep := range map[string]time.Duration{ResMinute: c.opts.RetainMinute, ResHour: c.opts.RetainHour, ResDay: c.opts.RetainDay} {
+		if keep <= 0 {
+			continue
+		}
+		if n, err := c.store.Prune(ctx, res, now-int64(keep.Seconds())); err != nil {
+			c.log.Warn("statistics: pruning failed", "err", err)
+		} else if n > 0 {
+			c.log.Debug("statistics: pruned", "resolution", res, "rows", n)
+		}
+	}
+}

internal/stats/collector_test.go

@@ -0,0 +1,238 @@
+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
+	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.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)
+
+	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])
+	}
+}

internal/stats/query.go

@@ -0,0 +1,480 @@
+package stats
+
+import (
+	"cmp"
+	"context"
+	"fmt"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Querier answers questions about stored statistics.
+type Querier struct {
+	Store *Store
+	Opts  Options
+	Now   func() time.Time
+}
+
+// NewQuerier returns a querier for a collector's data.
+func NewQuerier(c *Collector) *Querier {
+	return &Querier{Store: c.store, Opts: c.opts, Now: c.now}
+}
+
+func (q *Querier) loc() *time.Location {
+	if q.Opts.Location != nil {
+		return q.Opts.Location
+	}
+	return time.Local
+}
+
+// Resolution picks the finest stored resolution that covers a time range
+// starting since ago.
+func (q *Querier) Resolution(since time.Duration) string {
+	switch {
+	case since <= 48*time.Hour && since <= q.Opts.RetainMinute:
+		return ResMinute
+	case since <= q.Opts.RetainHour:
+		return ResHour
+	}
+	return ResDay
+}
+
+// topsResolution picks the resolution for top lists, which exist per hour
+// and per day.
+func (q *Querier) topsResolution(since time.Duration) string {
+	if since <= 7*24*time.Hour && since <= q.Opts.RetainHour {
+		return ResHour
+	}
+	return ResDay
+}
+
+// bucketStart returns the start of the bucket containing t at a resolution.
+func (q *Querier) bucketStart(res string, t time.Time) int64 {
+	switch res {
+	case ResMinute:
+		return t.Unix() - t.Unix()%60
+	case ResHour:
+		return t.Unix() - t.Unix()%3600
+	}
+	l := t.In(q.loc())
+	return time.Date(l.Year(), l.Month(), l.Day(), 0, 0, 0, 0, q.loc()).Unix()
+}
+
+// sums returns the totals of counters (optionally with a key prefix) in
+// [from, to).
+func (s *Store) sums(ctx context.Context, res string, from, to int64, prefix string) (map[string]int64, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT key, SUM(value) FROM counters WHERE res = ? AND ts >= ? AND ts < ? AND key LIKE ? ESCAPE '\' GROUP BY key`,
+		res, from, to, likePrefix(prefix))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[string]int64{}
+	for rows.Next() {
+		var k string
+		var v int64
+		if err := rows.Scan(&k, &v); err != nil {
+			return nil, err
+		}
+		out[k] = v
+	}
+	return out, rows.Err()
+}
+
+func likePrefix(p string) string {
+	r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
+	return r.Replace(p) + "%"
+}
+
+// series returns the values of a counter per bucket in [from, to).
+func (s *Store) series(ctx context.Context, res string, from, to int64, key string) (map[int64]int64, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT ts, value FROM counters WHERE res = ? AND key = ? AND ts >= ? AND ts < ?`, res, key, from, to)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[int64]int64{}
+	for rows.Next() {
+		var ts, v int64
+		if err := rows.Scan(&ts, &v); err != nil {
+			return nil, err
+		}
+		out[ts] = v
+	}
+	return out, rows.Err()
+}
+
+// tops sums top lists (of one or more kinds) in [from, to).
+func (s *Store) tops(ctx context.Context, res string, kinds []string, from, to int64, limit int) ([]Entry, error) {
+	args := []any{res}
+	for _, k := range kinds {
+		args = append(args, k)
+	}
+	args = append(args, from, to, limit)
+	rows, err := s.db.QueryContext(ctx, `SELECT key, SUM(count) AS n FROM tops WHERE res = ? AND kind IN (`+strings.TrimSuffix(strings.Repeat("?,", len(kinds)), ",")+`) AND ts >= ? AND ts < ?
+		GROUP BY key ORDER BY n DESC, key LIMIT ?`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []Entry{}
+	for rows.Next() {
+		var e Entry
+		if err := rows.Scan(&e.Key, &e.Count); err != nil {
+			return nil, err
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+func sorted(m map[string]int64, prefix string) []Entry {
+	out := []Entry{}
+	for k, v := range m {
+		if strings.HasPrefix(k, prefix) && v != 0 {
+			out = append(out, Entry{Key: strings.TrimPrefix(k, prefix), Count: v})
+		}
+	}
+	slices.SortFunc(out, func(a, b Entry) int {
+		if a.Count != b.Count {
+			return cmp.Compare(b.Count, a.Count)
+		}
+		return cmp.Compare(a.Key, b.Key)
+	})
+	return out
+}
+
+// Top returns the largest entries of a top list over the last since.
+func (q *Querier) Top(ctx context.Context, kind string, since time.Duration, limit int) ([]Entry, error) {
+	if !slices.Contains(Kinds, kind) && !quotaKind(kind) {
+		return nil, fmt.Errorf("unknown list %q (use %s, or quota_rN)", kind, strings.Join(Kinds, ", "))
+	}
+	now := q.Now()
+	res := q.topsResolution(since)
+	return q.Store.tops(ctx, res, []string{kind}, q.bucketStart(res, now.Add(-since)), now.Unix()+1, limit)
+}
+
+func quotaKind(k string) bool {
+	n, ok := strings.CutPrefix(k, "quota_r")
+	_, err := strconv.ParseInt(n, 10, 64)
+	return ok && err == nil
+}
+
+// Sums returns the totals of the counters with a prefix over the last
+// since, largest first, the prefix taken off ("hits:" → "r12").
+func (q *Querier) Sums(ctx context.Context, prefix string, since time.Duration) ([]Entry, error) {
+	now := q.Now()
+	res := q.Resolution(since)
+	m, err := q.Store.sums(ctx, res, q.bucketStart(res, now.Add(-since)), now.Unix()+1, prefix)
+	if err != nil {
+		return nil, err
+	}
+	return sorted(m, prefix), nil
+}
+
+// Traffic is what an interface or zone received and sent, in bytes.
+type Traffic struct {
+	Name     string `json:"name"`
+	Received int64  `json:"received"`
+	Sent     int64  `json:"sent"`
+}
+
+// Summary is an overview of the last since.
+type Summary struct {
+	Since      string    `json:"since"`
+	From       time.Time `json:"from"`
+	Zones      []Traffic `json:"zones"`
+	Interfaces []Traffic `json:"interfaces"`
+	// Drops are the packets dropped (rules, built-ins, policies), and by
+	// chain.
+	Drops        int64   `json:"drops"`
+	DropsByChain []Entry `json:"drops_by_chain"`
+	// Rules are the busiest rules, built-ins and policies, by packets.
+	Rules []Entry `json:"rules"`
+	// Connections: the most tracked at once (per-bucket averages), and
+	// the table's size.
+	PeakConnections int64   `json:"peak_connections"`
+	ConntrackMax    int64   `json:"conntrack_max"`
+	BlockedSources  []Entry `json:"blocked_sources"`
+	BlockedPorts    []Entry `json:"blocked_ports"`
+	DevicesUp       []Entry `json:"devices_up"`
+	DevicesDown     []Entry `json:"devices_down"`
+}
+
+// Summary sums up the last since, with top lists of up to topLimit.
+func (q *Querier) Summary(ctx context.Context, since time.Duration, topLimit int) (*Summary, error) {
+	now := q.Now()
+	res := q.Resolution(since)
+	from := q.bucketStart(res, now.Add(-since))
+	all, err := q.Store.sums(ctx, res, from, now.Unix()+1, "")
+	if err != nil {
+		return nil, err
+	}
+	s := &Summary{Since: humanDuration(since), From: time.Unix(from, 0), Zones: []Traffic{}, Interfaces: []Traffic{}}
+	traffic := func(rx, tx string) []Traffic {
+		by := map[string]*Traffic{}
+		for k, v := range all {
+			var name string
+			var in bool
+			if n, ok := strings.CutPrefix(k, rx); ok {
+				name, in = n, true
+			} else if n, ok := strings.CutPrefix(k, tx); ok {
+				name = n
+			} else {
+				continue
+			}
+			t := by[name]
+			if t == nil {
+				t = &Traffic{Name: name}
+				by[name] = t
+			}
+			if in {
+				t.Received += v
+			} else {
+				t.Sent += v
+			}
+		}
+		out := []Traffic{}
+		for _, t := range by {
+			out = append(out, *t)
+		}
+		slices.SortFunc(out, func(a, b Traffic) int {
+			if c := cmp.Compare(b.Received+b.Sent, a.Received+a.Sent); c != 0 {
+				return c
+			}
+			return cmp.Compare(a.Name, b.Name)
+		})
+		return out
+	}
+	s.Zones, s.Interfaces = traffic("zrx:", "ztx:"), traffic("rx:", "tx:")
+	s.Drops = all["drops"]
+	s.DropsByChain = sorted(all, "drop:")
+	s.Rules = sorted(all, "hits:")
+	if len(s.Rules) > topLimit {
+		s.Rules = s.Rules[:topLimit]
+	}
+	if tl, err := q.Timeline(ctx, since, "auto", []string{"g:" + GaugeConntrack, "g:" + GaugeConntrackMax}); err == nil {
+		for _, v := range tl.Series[0].Values {
+			s.PeakConnections = max(s.PeakConnections, v)
+		}
+		for _, v := range tl.Series[1].Values {
+			s.ConntrackMax = max(s.ConntrackMax, v)
+		}
+	}
+	for _, l := range []struct {
+		kind string
+		dst  *[]Entry
+	}{{KindBlockedSource, &s.BlockedSources}, {KindBlockedPort, &s.BlockedPorts}, {KindDeviceUp, &s.DevicesUp}, {KindDeviceDown, &s.DevicesDown}} {
+		es, err := q.Top(ctx, l.kind, since, topLimit)
+		if err != nil {
+			return nil, err
+		}
+		*l.dst = es
+	}
+	return s, nil
+}
+
+// Timeline is a set of series over time.
+type Timeline struct {
+	Step   string    `json:"step"`
+	Times  []int64   `json:"times"` // bucket starts (Unix seconds)
+	Series []*Series `json:"series"`
+}
+
+// Series is one counter over time. For a gauge (g:KEY) the values are
+// averages, and -1 where there is no sample.
+type Series struct {
+	Key    string  `json:"key"`
+	Values []int64 `json:"values"`
+}
+
+var steps = []struct {
+	name string
+	d    time.Duration
+	res  string
+}{
+	{"1m", time.Minute, ResMinute}, {"5m", 5 * time.Minute, ResMinute}, {"15m", 15 * time.Minute, ResMinute},
+	{"1h", time.Hour, ResHour}, {"6h", 6 * time.Hour, ResHour}, {"1d", 24 * time.Hour, ResDay},
+}
+
+// Timeline returns counters over the last since. keys are counter names;
+// a key ending in ":*" means every counter of that group. step is "auto"
+// or one of 1m, 5m, 15m, 1h, 6h, 1d.
+func (q *Querier) Timeline(ctx context.Context, since time.Duration, step string, keys []string) (*Timeline, error) {
+	idx := -1
+	for i, st := range steps {
+		ok := st.res == ResMinute && since <= q.Opts.RetainMinute || st.res == ResHour && since <= q.Opts.RetainHour || st.res == ResDay
+		if step == st.name {
+			if !ok {
+				return nil, fmt.Errorf("step %s needs %s data, which is only kept for %s", st.name, map[string]string{ResMinute: "per-minute", ResHour: "hourly"}[st.res], humanDuration(map[string]time.Duration{ResMinute: q.Opts.RetainMinute, ResHour: q.Opts.RetainHour}[st.res]))
+			}
+			idx = i
+			break
+		}
+		if step == "auto" || step == "" {
+			if ok && since/st.d <= 300 {
+				idx = i
+				break
+			}
+		}
+	}
+	if idx < 0 {
+		if step != "auto" && step != "" {
+			return nil, fmt.Errorf("unknown step %q (use auto, 1m, 5m, 15m, 1h, 6h or 1d)", step)
+		}
+		idx = len(steps) - 1
+	}
+	st := steps[idx]
+	now := q.Now()
+	// Buckets aligned to the step (days to local midnight).
+	var starts []int64
+	if st.res == ResDay {
+		for t := time.Unix(q.bucketStart(ResDay, now.Add(-since)), 0).In(q.loc()); !t.After(now); t = t.AddDate(0, 0, 1) {
+			starts = append(starts, t.Unix())
+		}
+	} else {
+		sd := int64(st.d.Seconds())
+		first := now.Add(-since).Unix()
+		first -= first % sd
+		for t := first; t <= now.Unix(); t += sd {
+			starts = append(starts, t)
+		}
+	}
+	from, to := starts[0], now.Unix()+1
+	// Expand group keys.
+	var expanded []string
+	for _, k := range keys {
+		if g, ok := strings.CutSuffix(k, ":*"); ok {
+			m, err := q.Store.sums(ctx, st.res, from, to, g+":")
+			if err != nil {
+				return nil, err
+			}
+			for _, e := range sorted(m, "") {
+				expanded = append(expanded, e.Key)
+			}
+			continue
+		}
+		expanded = append(expanded, k)
+	}
+	tl := &Timeline{Step: st.name, Times: starts, Series: []*Series{}}
+	bucketed := func(key string) ([]int64, error) {
+		raw, err := q.Store.series(ctx, st.res, from, to, key)
+		if err != nil {
+			return nil, err
+		}
+		vals := make([]int64, len(starts))
+		for ts, v := range raw {
+			// The bucket is the last start <= ts.
+			i, found := slices.BinarySearch(starts, ts)
+			if !found {
+				i--
+			}
+			if i >= 0 && i < len(vals) {
+				vals[i] += v
+			}
+		}
+		return vals, nil
+	}
+	for _, k := range expanded {
+		vals, err := bucketed(k)
+		if err != nil {
+			return nil, err
+		}
+		// A gauge (g:KEY) is the average of its samples: its sum divided
+		// by gn:KEY, their number. A bucket without samples is -1.
+		if g, ok := strings.CutPrefix(k, "g:"); ok {
+			n, err := bucketed("gn:" + g)
+			if err != nil {
+				return nil, err
+			}
+			for i := range vals {
+				if n[i] > 0 {
+					vals[i] /= n[i]
+				} else {
+					vals[i] = -1
+				}
+			}
+		}
+		tl.Series = append(tl.Series, &Series{Key: k, Values: vals})
+	}
+	return tl, nil
+}
+
+// Heatmap is average activity per weekday and hour of day (local time).
+type Heatmap struct {
+	Key string `json:"key"`
+	// Avg[d][h] is the average per hour on weekday d (0 = Monday) at
+	// hour h; -1 where there is no data.
+	Avg [7][24]float64 `json:"avg"`
+	// Samples[d][h] is how many hours the average is based on.
+	Samples [7][24]int `json:"samples"`
+	From    time.Time  `json:"from"`
+}
+
+// Heatmap averages a counter (default "drops") by weekday and hour over
+// the last since (at most the hourly retention).
+func (q *Querier) Heatmap(ctx context.Context, key string, since time.Duration) (*Heatmap, error) {
+	if key == "" {
+		key = "drops"
+	}
+	since = min(since, q.Opts.RetainHour)
+	now := q.Now()
+	from := q.bucketStart(ResHour, now.Add(-since))
+	raw, err := q.Store.series(ctx, ResHour, from, now.Unix()+1, key)
+	if err != nil {
+		return nil, err
+	}
+	hm := &Heatmap{Key: key, From: time.Unix(from, 0)}
+	var sum [7][24]int64
+	for ts, v := range raw {
+		t := time.Unix(ts, 0).In(q.loc())
+		d := (int(t.Weekday()) + 6) % 7
+		sum[d][t.Hour()] += v
+		hm.Samples[d][t.Hour()]++
+	}
+	for d := range 7 {
+		for h := range 24 {
+			if n := hm.Samples[d][h]; n > 0 {
+				hm.Avg[d][h] = float64(sum[d][h]) / float64(n)
+			} else {
+				hm.Avg[d][h] = -1
+			}
+		}
+	}
+	return hm, nil
+}
+
+// ParseSince parses a range such as 90m, 24h, 7d or 2w.
+func ParseSince(s string) (time.Duration, error) {
+	if s == "" {
+		return 24 * time.Hour, nil
+	}
+	mult := time.Duration(0)
+	switch {
+	case strings.HasSuffix(s, "d"):
+		mult = 24 * time.Hour
+	case strings.HasSuffix(s, "w"):
+		mult = 7 * 24 * time.Hour
+	}
+	if mult > 0 {
+		var n int
+		if _, err := fmt.Sscanf(s[:len(s)-1], "%d", &n); err != nil || n <= 0 || fmt.Sprint(n) != s[:len(s)-1] {
+			return 0, fmt.Errorf("invalid range %q (use e.g. 90m, 24h, 7d, 2w)", s)
+		}
+		return time.Duration(n) * mult, nil
+	}
+	d, err := time.ParseDuration(s)
+	if err != nil || d <= 0 {
+		return 0, fmt.Errorf("invalid range %q (use e.g. 90m, 24h, 7d, 2w)", s)
+	}
+	return d, nil
+}
+
+func humanDuration(d time.Duration) string {
+	if d%(24*time.Hour) == 0 && d > 0 {
+		return strconv.Itoa(int(d/(24*time.Hour))) + "d"
+	}
+	return d.String()
+}

internal/stats/sources.go

@@ -0,0 +1,136 @@
+package stats
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+)
+
+// What the collector reads from the kernel without privileges: the
+// interfaces' counters (/sys/class/net), the connection tracking table's
+// size, and housecarld's and the host's memory and load (/proc).
+
+// clockTicks is USER_HZ, the unit of the CPU times in /proc/PID/stat. It
+// is 100 on every Linux architecture Housecarl runs on.
+const clockTicks = 100
+
+// ifCounters are an interface's totals since it came up.
+type ifCounters struct {
+	RxBytes, TxBytes, RxPackets, TxPackets int64
+}
+
+// readInterfaces reads every interface's counters but the loopback's.
+func readInterfaces(sysNet string) map[string]ifCounters {
+	out := map[string]ifCounters{}
+	entries, err := os.ReadDir(sysNet)
+	if err != nil {
+		return out
+	}
+	for _, e := range entries {
+		name := e.Name()
+		if name == "lo" {
+			continue
+		}
+		dir := filepath.Join(sysNet, name, "statistics")
+		read := func(f string) (int64, bool) {
+			b, err := os.ReadFile(filepath.Join(dir, f))
+			if err != nil {
+				return 0, false
+			}
+			n, err := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
+			return n, err == nil
+		}
+		var c ifCounters
+		var ok bool
+		if c.RxBytes, ok = read("rx_bytes"); !ok {
+			continue
+		}
+		c.TxBytes, _ = read("tx_bytes")
+		c.RxPackets, _ = read("rx_packets")
+		c.TxPackets, _ = read("tx_packets")
+		out[name] = c
+	}
+	return out
+}
+
+// readConntrack returns how many connections the kernel tracks, and how
+// many it can (ok false without connection tracking).
+func readConntrack(proc string) (count, max int64, ok bool) {
+	read := func(f string) (int64, bool) {
+		b, err := os.ReadFile(filepath.Join(proc, "sys/net/netfilter", f))
+		if err != nil {
+			return 0, false
+		}
+		n, err := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
+		return n, err == nil
+	}
+	count, ok1 := read("nf_conntrack_count")
+	max, ok2 := read("nf_conntrack_max")
+	return count, max, ok1 && ok2
+}
+
+// readSelf returns housecarld's resident memory (bytes) and CPU time
+// (clock ticks).
+func readSelf(proc string) (rss, cpu int64, ok bool) {
+	dir := filepath.Join(proc, "self")
+	stat, err := os.ReadFile(filepath.Join(dir, "stat"))
+	if err != nil {
+		return 0, 0, false
+	}
+	// The command name is in parentheses and may contain spaces: the
+	// fields that follow start after the last ')'.
+	i := bytes.LastIndexByte(stat, ')')
+	if i < 0 {
+		return 0, 0, false
+	}
+	f := strings.Fields(string(stat[i+1:]))
+	// f[0] is field 3 (state); utime and stime are fields 14 and 15.
+	if len(f) < 13 {
+		return 0, 0, false
+	}
+	utime, _ := strconv.ParseInt(f[11], 10, 64)
+	stime, _ := strconv.ParseInt(f[12], 10, 64)
+	status, err := os.ReadFile(filepath.Join(dir, "status"))
+	if err != nil {
+		return 0, 0, false
+	}
+	for _, line := range strings.Split(string(status), "\n") {
+		if k, v, ok := strings.Cut(line, ":"); ok && k == "VmRSS" {
+			rss = kib(v)
+		}
+	}
+	return rss, utime + stime, true
+}
+
+// readHost returns the memory in use (bytes) and the 1-minute load
+// average.
+func readHost(proc string) (memUsed int64, load float64, ok bool) {
+	b, err := os.ReadFile(filepath.Join(proc, "meminfo"))
+	if err != nil {
+		return 0, 0, false
+	}
+	var total, avail int64
+	for _, line := range strings.Split(string(b), "\n") {
+		k, v, _ := strings.Cut(line, ":")
+		switch k {
+		case "MemTotal":
+			total = kib(v)
+		case "MemAvailable":
+			avail = kib(v)
+		}
+	}
+	if b, err := os.ReadFile(filepath.Join(proc, "loadavg")); err == nil {
+		if f := strings.Fields(string(b)); len(f) > 0 {
+			load, _ = strconv.ParseFloat(f[0], 64)
+		}
+	}
+	return total - avail, load, total > 0
+}
+
+// kib parses a "1234 kB" value from /proc into bytes.
+func kib(v string) int64 {
+	n, _ := strconv.ParseInt(strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(v), "kB")), 10, 64)
+	return n * 1024
+}

internal/stats/store.go

@@ -0,0 +1,304 @@
+// Package stats keeps the firewall's figures over time: traffic per
+// interface and zone, hits per rule, drops, blocked sources, connections,
+// quotas and the machine's health, sampled every minute. It stores
+// aggregates in a SQLite database of its own, separate from Housecarl's
+// configuration, so it can be deleted at any time without losing anything
+// important.
+package stats
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+
+	_ "modernc.org/sqlite" // registers the "sqlite" driver
+)
+
+// Resolutions of stored data.
+const (
+	ResMinute = "m"
+	ResHour   = "h"
+	ResDay    = "d"
+)
+
+// Store is the statistics database.
+type Store struct {
+	db   *sql.DB
+	path string
+}
+
+var schema = []string{
+	`CREATE TABLE counters (
+		res   TEXT NOT NULL,
+		ts    INTEGER NOT NULL,
+		key   TEXT NOT NULL,
+		value INTEGER NOT NULL,
+		PRIMARY KEY (res, ts, key)
+	) WITHOUT ROWID;
+	CREATE INDEX counters_by_key ON counters (res, key, ts);
+	CREATE TABLE tops (
+		res   TEXT NOT NULL,
+		ts    INTEGER NOT NULL,
+		kind  TEXT NOT NULL,
+		key   TEXT NOT NULL,
+		count INTEGER NOT NULL,
+		PRIMARY KEY (res, ts, kind, key)
+	) WITHOUT ROWID;
+	CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);`,
+}
+
+// Size is the database's size on disk, with its write-ahead log.
+func (s *Store) Size() int64 {
+	var n int64
+	for _, f := range []string{s.path, s.path + "-wal"} {
+		if fi, err := os.Stat(f); err == nil {
+			n += fi.Size()
+		}
+	}
+	return n
+}
+
+// Oldest is the start of the oldest stored bucket (0 when empty).
+func (s *Store) Oldest(ctx context.Context) (int64, error) {
+	var ts sql.NullInt64
+	err := s.db.QueryRowContext(ctx, `SELECT MIN(ts) FROM counters`).Scan(&ts)
+	return ts.Int64, err
+}
+
+// OpenStore opens (creating if needed) the statistics database. Use
+// ":memory:" for tests.
+func OpenStore(path string) (*Store, error) {
+	dsn := "file:" + path + "?_pragma=busy_timeout(10000)&_pragma=synchronous(NORMAL)&_txlock=immediate"
+	if path == ":memory:" {
+		dsn = "file::memory:?_txlock=immediate"
+	} else {
+		dsn += "&_pragma=journal_mode(WAL)"
+	}
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, err
+	}
+	if path == ":memory:" {
+		db.SetMaxOpenConns(1)
+	}
+	s := &Store{db: db, path: path}
+	if err := s.migrate(); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("statistics database %s: %w", path, err)
+	}
+	return s, nil
+}
+
+// Path returns the database file ("" in memory).
+func (s *Store) Path() string {
+	if s.path == ":memory:" {
+		return ""
+	}
+	return s.path
+}
+
+// Close closes the database.
+func (s *Store) Close() error { return s.db.Close() }
+
+func (s *Store) migrate() error {
+	if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)`); err != nil {
+		return err
+	}
+	var v int
+	err := s.db.QueryRow(`SELECT version FROM schema_version`).Scan(&v)
+	if errors.Is(err, sql.ErrNoRows) {
+		if _, err := s.db.Exec(`INSERT INTO schema_version (version) VALUES (0)`); err != nil {
+			return err
+		}
+	} else if err != nil {
+		return err
+	}
+	if v > len(schema) {
+		return fmt.Errorf("schema version %d is newer than this Housecarl; upgrade Housecarl or delete the statistics database", v)
+	}
+	for i := v; i < len(schema); i++ {
+		tx, err := s.db.Begin()
+		if err != nil {
+			return err
+		}
+		if _, err := tx.Exec(schema[i]); err != nil {
+			tx.Rollback()
+			return err
+		}
+		if _, err := tx.Exec(`UPDATE schema_version SET version = ?`, i+1); err != nil {
+			tx.Rollback()
+			return err
+		}
+		if err := tx.Commit(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Batch collects writes that are committed together.
+type Batch struct {
+	counters []counterRow
+	tops     []topsWrite
+	meta     map[string]string
+}
+
+type counterRow struct {
+	res   string
+	ts    int64
+	key   string
+	value int64
+}
+
+type topsWrite struct {
+	res, kind string
+	ts        int64
+	entries   []Entry
+	replace   bool // replace the (res, ts, kind) rows instead of adding
+	keep      int  // after adding, keep only the largest entries (0: all)
+}
+
+// Entry is a key and its count.
+type Entry struct {
+	Key   string `json:"key"`
+	Count int64  `json:"count"`
+	// Name is what the key stands for (a rule's label, a device's host
+	// name), set by the API.
+	Name string `json:"name,omitempty"`
+}
+
+// AddCounters adds values to the counters of a bucket.
+func (b *Batch) AddCounters(res string, ts int64, values map[string]int64) {
+	for k, v := range values {
+		b.counters = append(b.counters, counterRow{res, ts, k, v})
+	}
+}
+
+// SetTops replaces a bucket's top list of a kind.
+func (b *Batch) SetTops(res string, ts int64, kind string, entries []Entry) {
+	b.tops = append(b.tops, topsWrite{res: res, ts: ts, kind: kind, entries: entries, replace: true})
+}
+
+// AddTops adds counts to a bucket's top list and keeps the keep largest.
+func (b *Batch) AddTops(res string, ts int64, kind string, entries []Entry, keep int) {
+	b.tops = append(b.tops, topsWrite{res: res, ts: ts, kind: kind, entries: entries, keep: keep})
+}
+
+// SetMeta stores a value that survives restarts.
+func (b *Batch) SetMeta(key, value string) {
+	if b.meta == nil {
+		b.meta = map[string]string{}
+	}
+	b.meta[key] = value
+}
+
+// Commit writes the batch in one transaction.
+func (s *Store) Commit(ctx context.Context, b *Batch) error {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if len(b.counters) > 0 {
+		st, err := tx.PrepareContext(ctx, `INSERT INTO counters (res, ts, key, value) VALUES (?, ?, ?, ?)
+			ON CONFLICT (res, ts, key) DO UPDATE SET value = value + excluded.value`)
+		if err != nil {
+			return err
+		}
+		for _, c := range b.counters {
+			if _, err := st.ExecContext(ctx, c.res, c.ts, c.key, c.value); err != nil {
+				st.Close()
+				return err
+			}
+		}
+		st.Close()
+	}
+	for _, t := range b.tops {
+		if t.replace {
+			if _, err := tx.ExecContext(ctx, `DELETE FROM tops WHERE res = ? AND ts = ? AND kind = ?`, t.res, t.ts, t.kind); err != nil {
+				return err
+			}
+		}
+		for _, e := range t.entries {
+			if _, err := tx.ExecContext(ctx, `INSERT INTO tops (res, ts, kind, key, count) VALUES (?, ?, ?, ?, ?)
+				ON CONFLICT (res, ts, kind, key) DO UPDATE SET count = count + excluded.count`, t.res, t.ts, t.kind, e.Key, e.Count); err != nil {
+				return err
+			}
+		}
+		if t.keep > 0 {
+			if _, err := tx.ExecContext(ctx, `DELETE FROM tops WHERE res = ? AND ts = ? AND kind = ? AND key NOT IN (
+				SELECT key FROM tops WHERE res = ? AND ts = ? AND kind = ? ORDER BY count DESC, key LIMIT ?)`,
+				t.res, t.ts, t.kind, t.res, t.ts, t.kind, t.keep); err != nil {
+				return err
+			}
+		}
+	}
+	for k, v := range b.meta {
+		if _, err := tx.ExecContext(ctx, `INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value`, k, v); err != nil {
+			return err
+		}
+	}
+	return tx.Commit()
+}
+
+// Meta returns a stored value ("" if unset).
+func (s *Store) Meta(ctx context.Context, key string) (string, error) {
+	var v string
+	err := s.db.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = ?`, key).Scan(&v)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", nil
+	}
+	return v, err
+}
+
+// Prune deletes data of a resolution older than before (a Unix time).
+func (s *Store) Prune(ctx context.Context, res string, before int64) (int64, error) {
+	var n int64
+	for _, table := range []string{"counters", "tops"} {
+		r, err := s.db.ExecContext(ctx, `DELETE FROM `+table+` WHERE res = ? AND ts < ?`, res, before)
+		if err != nil {
+			return n, err
+		}
+		c, _ := r.RowsAffected()
+		n += c
+	}
+	return n, nil
+}
+
+// TopsAt returns a bucket's top list of a kind.
+func (s *Store) TopsAt(ctx context.Context, res string, ts int64, kind string) ([]Entry, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT key, count FROM tops WHERE res = ? AND ts = ? AND kind = ? ORDER BY count DESC, key`, res, ts, kind)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Entry
+	for rows.Next() {
+		var e Entry
+		if err := rows.Scan(&e.Key, &e.Count); err != nil {
+			return nil, err
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+// KindsAt returns the top-list kinds stored for one bucket.
+func (s *Store) KindsAt(ctx context.Context, res string, ts int64) ([]string, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT kind FROM tops WHERE res = ? AND ts = ?`, res, ts)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var k string
+		if err := rows.Scan(&k); err != nil {
+			return nil, err
+		}
+		out = append(out, k)
+	}
+	return out, rows.Err()
+}