Add optional query statistics

Off by default (stats in jarl.yaml). When enabled, Jarl adds a loopback
statistics channel and a response log to named's configuration (merged
into existing logging/statistics-channels statements) and switches
response logging with rndc responselog, because the responselog option
only counts when named starts. jarld polls named's counters every minute
and tails the response log (resuming after restarts and across
rotation). It stores aggregates, never individual queries, in a separate
SQLite database: per-minute, hourly and daily counters (queries, result
codes, record types, views, zones, cache hits, upstream latency, DNSSEC)
and hourly/daily top lists (names, clients, NXDOMAIN and SERVFAIL names,
refused clients), each with its own retention. Clients can be anonymized
or excluded.

- Web UI: Statistics page with KPI tiles, queries and failed answers over
  time, a weekday x hour heatmap, top lists and breakdowns; SVG charts
  with hover/keyboard readouts, a table view and light/dark palettes
  validated for colour-vision deficiency.
- CLI: jarl stats status|summary|top|timeline|heatmap|breakdown.
- API: /api/v1/stats/*, behind a new "stats" permission resource.
- jarld brings named's configuration up to date with jarl.yaml at
  startup (engine.SyncConfig), never overwriting hand-edited files.
- resync now refuses to overwrite drifted files without force, as
  documented.
- tmpfiles.d creates /var/log/named (allowed by named's AppArmor profile
  on Debian 13 and Ubuntu).
- The e2e harness runs named with -f like the Debian unit.
- docs/stats.md, plus API, CLI, configuration, permissions and README
  updates.

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

internal/stats/collector.go

@@ -0,0 +1,564 @@
+package stats
+
+import (
+	"cmp"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/netip"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Kinds of top lists.
+const (
+	KindName     = "name"     // most queried names
+	KindClient   = "client"   // clients sending the most queries
+	KindNXDomain = "nxdomain" // names answered with NXDOMAIN
+	KindServFail = "servfail" // names answered with SERVFAIL
+	KindRefused  = "refused"  // clients that were refused
+)
+
+// Kinds lists the top-list kinds.
+var Kinds = []string{KindName, KindClient, KindNXDomain, KindServFail, KindRefused}
+
+// OtherZone is the zone key for names outside every local zone (answered
+// by recursion, or refused).
+const OtherZone = "-"
+
+// Options configure a Collector.
+type Options struct {
+	QueryLog          bool
+	LogFile           string
+	StatisticsChannel string // ADDRESS:PORT
+	TopN              int
+	RetainMinute      time.Duration
+	RetainHour        time.Duration
+	RetainDay         time.Duration
+	Anonymize         bool
+	Exclude           []netip.Prefix
+	Location          *time.Location // for days and log timestamps (default: local time)
+	// EnsureLogging turns named's response logging on (rndc responselog
+	// on). It is called at start and when named answers queries but no
+	// log lines arrive.
+	EnsureLogging func(context.Context) (bool, error)
+}
+
+// ZoneIndex returns, per view name ("" when no views are used), the names
+// of the zones this server is authoritative for (lowercase, without the
+// final dot).
+type ZoneIndex func() map[string][]string
+
+// Status describes the collector's health.
+type Status struct {
+	Enabled           bool       `json:"enabled"`
+	QueryLog          bool       `json:"query_log"`
+	LogFile           string     `json:"log_file,omitempty"`
+	StatisticsChannel string     `json:"statistics_channel"`
+	LastPoll          *time.Time `json:"last_poll,omitempty"`
+	PollError         string     `json:"poll_error,omitempty"`
+	LastLine          *time.Time `json:"last_log_line,omitempty"`
+	LinesRead         int64      `json:"log_lines_read"`
+	LogError          string     `json:"log_error,omitempty"`
+	Warnings          []string   `json:"warnings"`
+	FirstData         *time.Time `json:"first_data,omitempty"`
+	Retention         struct {
+		Minute string `json:"minute"`
+		Hour   string `json:"hour"`
+		Day    string `json:"day"`
+	} `json:"retention"`
+}
+
+// Collector gathers statistics into a Store.
+type Collector struct {
+	opts   Options
+	store  *Store
+	zones  ZoneIndex
+	log    *slog.Logger
+	poller *Poller
+	tail   *Tailer
+	now    func() time.Time
+
+	mu      sync.Mutex
+	minutes map[int64]map[string]int64            // minute -> counter -> value
+	hours   map[int64]map[string]map[string]int64 // live hours: hour -> kind -> key -> count
+	// late holds counts for hours that were already written and added to
+	// their day (lines read after a restart); they are added, not
+	// replaced.
+	late     map[int64]map[string]map[string]int64
+	liveHour int64
+	prev     *Snapshot
+	zoneIdx  map[string][]string
+	zoneAt   time.Time
+	status   Status
+	recentQ  []int64 // queries per minute from the counters, last 10 minutes
+	recentL  []int64 // response log lines per minute, last 10 minutes
+	linesMin int64
+	ensureAt time.Time
+}
+
+// maxKeys bounds the per-hour maps; rarely seen keys are dropped first.
+const maxKeys = 200_000
+
+// NewCollector creates a collector. Run starts it.
+func NewCollector(store *Store, opts Options, zones ZoneIndex, log *slog.Logger) *Collector {
+	if opts.Location == nil {
+		opts.Location = time.Local
+	}
+	c := &Collector{opts: opts, store: store, zones: zones, log: log, poller: NewPoller(opts.StatisticsChannel), now: time.Now,
+		minutes: map[int64]map[string]int64{}, hours: map[int64]map[string]map[string]int64{}, late: map[int64]map[string]map[string]int64{}}
+	if opts.QueryLog {
+		c.tail = &Tailer{Path: opts.LogFile}
+	}
+	c.status = Status{Enabled: true, QueryLog: opts.QueryLog, StatisticsChannel: opts.StatisticsChannel, Warnings: []string{}}
+	if opts.QueryLog {
+		c.status.LogFile = opts.LogFile
+	}
+	c.status.Retention.Minute = humanDuration(opts.RetainMinute)
+	c.status.Retention.Hour = humanDuration(opts.RetainHour)
+	c.status.Retention.Day = humanDuration(opts.RetainDay)
+	return c
+}
+
+// Store returns the collector's store.
+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 current status.
+func (c *Collector) Status() Status {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	s := c.status
+	s.Warnings = append([]string{}, c.status.Warnings...)
+	return s
+}
+
+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)
+	tick := time.NewTicker(time.Second)
+	defer tick.Stop()
+	next := c.nextFlush()
+	for {
+		select {
+		case <-ctx.Done():
+			// Write what was collected; a restart continues the minute.
+			c.readLog()
+			c.flush(context.Background(), true)
+			if c.tail != nil {
+				c.tail.Close()
+			}
+			return
+		case <-tick.C:
+			c.readLog()
+			if !c.now().Before(next) {
+				c.flush(ctx, false)
+				next = c.nextFlush()
+			}
+		}
+	}
+}
+
+// nextFlush is two seconds into the next minute, so the minute that just
+// ended is complete.
+func (c *Collector) nextFlush() time.Time {
+	return time.Unix(c.minuteOf(c.now())+62, 0)
+}
+
+func (c *Collector) start(ctx context.Context) {
+	if c.tail != nil {
+		if v, _ := c.store.Meta(ctx, "tail"); v != "" {
+			var p Position
+			if json.Unmarshal([]byte(v), &p) == nil {
+				c.tail.Resume(p)
+			}
+		}
+	}
+	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
+		}
+	}
+	// Continue the current hour's top lists, and fold hours that ended
+	// while jarld was not running into their days.
+	cur := c.hourOf(c.now())
+	c.liveHour = cur
+	b := &Batch{}
+	merged, _ := c.store.Meta(ctx, "merged_through")
+	if last, err := strconv.ParseInt(merged, 10, 64); err == nil {
+		from := max(last+3600, cur-48*3600)
+		for h := from; h < cur; h += 3600 {
+			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))
+	for _, k := range Kinds {
+		es, err := c.store.TopsAt(ctx, ResHour, cur, k)
+		if err != nil || len(es) == 0 {
+			continue
+		}
+		m := c.topsMap(c.hours, cur, k)
+		for _, e := range es {
+			m[e.Key] = e.Count
+		}
+	}
+	if err := c.store.Commit(ctx, b); err != nil {
+		c.log.Warn("statistics: catching up failed", "err", err)
+	}
+	c.ensureLogging(ctx)
+	// The first poll is the baseline for the counters.
+	if snap, err := c.poller.Poll(ctx); err == nil {
+		c.prev = snap
+		t := c.now()
+		c.status.LastPoll = &t
+	} else {
+		c.status.PollError = err.Error()
+	}
+	c.prune(ctx)
+	c.log.Info("statistics collector started", "query_log", c.opts.QueryLog, "statistics_channel", c.opts.StatisticsChannel)
+}
+
+func (c *Collector) topsMap(hours map[int64]map[string]map[string]int64, hour int64, kind string) map[string]int64 {
+	h := hours[hour]
+	if h == nil {
+		h = map[string]map[string]int64{}
+		hours[hour] = h
+	}
+	m := h[kind]
+	if m == nil {
+		m = map[string]int64{}
+		h[kind] = m
+	}
+	return m
+}
+
+// hourMaps returns the per-kind maps of an hour, creating them.
+func (c *Collector) hourMaps(hours map[int64]map[string]map[string]int64, hour int64) map[string]map[string]int64 {
+	for _, k := range Kinds {
+		c.topsMap(hours, hour, k)
+	}
+	return hours[hour]
+}
+
+func (c *Collector) count(minute int64, key string, n int64) {
+	m := c.minutes[minute]
+	if m == nil {
+		m = map[string]int64{}
+		c.minutes[minute] = m
+	}
+	m[key] += n
+}
+
+// readLog reads new response log lines.
+func (c *Collector) readLog() {
+	if c.tail == nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.refreshZones()
+	_, err := c.tail.Read(c.handleLine)
+	if err != nil {
+		c.status.LogError = err.Error()
+	} else {
+		c.status.LogError = ""
+	}
+}
+
+// HandleLine processes one log line (exported for tests).
+func (c *Collector) HandleLine(line string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.refreshZones()
+	c.handleLine(line)
+}
+
+func (c *Collector) handleLine(line string) {
+	ev, ok := ParseResponse(line, c.opts.Location)
+	if !ok {
+		return
+	}
+	c.status.LinesRead++
+	c.linesMin++
+	t := ev.Time
+	c.status.LastLine = &t
+	for _, p := range c.opts.Exclude {
+		if p.Contains(ev.Client) {
+			return
+		}
+	}
+	minute, hour := c.minuteOf(ev.Time), c.hourOf(ev.Time)
+	view := ev.View
+	if view == "" {
+		view = "_default"
+	}
+	c.count(minute, "responses", 1)
+	c.count(minute, "view:"+view, 1)
+	c.count(minute, "zone:"+c.zoneOf(ev.Name, ev.View), 1)
+	client := clientKey(ev.Client, c.opts.Anonymize)
+	hours := c.hours
+	if hour < c.liveHour {
+		hours = c.late
+	}
+	hours[hour] = c.hourMaps(hours, hour)
+	tops := hours[hour]
+	tops[KindName][ev.Name]++
+	tops[KindClient][client]++
+	switch ev.RCode {
+	case "NXDOMAIN":
+		tops[KindNXDomain][ev.Name]++
+	case "SERVFAIL":
+		tops[KindServFail][ev.Name]++
+	case "REFUSED":
+		tops[KindRefused][client]++
+	}
+	for _, m := range tops {
+		if len(m) > maxKeys {
+			shrink(m)
+		}
+	}
+}
+
+// shrink drops the rarest keys until the map is well under maxKeys.
+func shrink(m map[string]int64) {
+	for limit := int64(1); len(m) > maxKeys*8/10; limit++ {
+		for k, v := range m {
+			if v <= limit {
+				delete(m, k)
+			}
+		}
+	}
+}
+
+// clientKey is how a client is recorded: its address, or its network
+// when anonymizing.
+func clientKey(a netip.Addr, anonymize bool) string {
+	if !anonymize {
+		return a.String()
+	}
+	bits := 24
+	if a.Is6() {
+		bits = 48
+	}
+	p, _ := a.Prefix(bits)
+	return p.String()
+}
+
+func (c *Collector) refreshZones() {
+	if c.zones == nil || c.zoneIdx != nil && c.now().Sub(c.zoneAt) < 5*time.Minute {
+		return
+	}
+	idx := c.zones()
+	for v, zs := range idx {
+		// Longest first, so the most specific zone matches.
+		slices.SortFunc(zs, func(a, b string) int { return cmp.Compare(len(b), len(a)) })
+		idx[v] = zs
+	}
+	c.zoneIdx, c.zoneAt = idx, c.now()
+}
+
+// zoneOf returns the local zone a name belongs to in a view, or OtherZone.
+func (c *Collector) zoneOf(name, view string) string {
+	for _, z := range c.zoneIdx[view] {
+		if name == z || z == "." || strings.HasSuffix(name, "."+z) {
+			return z
+		}
+	}
+	return OtherZone
+}
+
+// flush polls named's counters and writes everything collected for
+// finished minutes (all minutes when final).
+func (c *Collector) flush(ctx context.Context, final bool) {
+	now := c.now()
+	cur := c.minuteOf(now)
+	snap, perr := c.poller.Poll(ctx)
+
+	c.mu.Lock()
+	if perr != nil {
+		c.status.PollError = perr.Error()
+	} else {
+		c.status.PollError = ""
+		c.status.LastPoll = &now
+		if c.prev != nil {
+			d := Delta(c.prev, snap)
+			// The change happened during the minute that just ended.
+			for k, v := range d {
+				c.count(cur-60, k, v)
+			}
+			c.noteRecent(d["queries"])
+		}
+		c.prev = snap
+	}
+	b := &Batch{}
+	first := c.status.FirstData == nil
+	for m, vals := range c.minutes {
+		if m >= cur && !final {
+			continue
+		}
+		t := time.Unix(m, 0)
+		b.AddCounters(ResMinute, m, vals)
+		b.AddCounters(ResHour, c.hourOf(t), vals)
+		b.AddCounters(ResDay, c.dayOf(t), vals)
+		delete(c.minutes, m)
+		if first && len(vals) > 0 {
+			ft := t
+			c.status.FirstData = &ft
+			first = false
+			b.SetMeta("first_data", strconv.FormatInt(m, 10))
+		}
+	}
+	curHour := c.hourOf(now)
+	for h, kinds := range c.late {
+		for k, m := range kinds {
+			if len(m) > 0 {
+				b.AddTops(ResHour, h, k, top(m, c.opts.TopN), c.opts.TopN)
+				b.AddTops(ResDay, c.dayOf(time.Unix(h, 0)), k, top(m, c.opts.TopN), 3*c.opts.TopN)
+			}
+		}
+		delete(c.late, h)
+	}
+	for h, kinds := range c.hours {
+		for k, m := range kinds {
+			b.SetTops(ResHour, h, k, top(m, c.opts.TopN))
+		}
+		if h < curHour {
+			// 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.liveHour = curHour
+	if c.tail != nil {
+		pos, _ := json.Marshal(c.tail.Position())
+		b.SetMeta("tail", string(pos))
+	}
+	c.updateWarnings()
+	c.mu.Unlock()
+
+	if err := c.store.Commit(ctx, b); err != nil {
+		c.log.Error("statistics: writing failed", "err", err)
+	}
+	// Queries but no log lines for a while: named's response logging may
+	// have been turned off (or named restarted without it).
+	c.mu.Lock()
+	quiet := len(c.recentL) >= 2 && c.recentL[len(c.recentL)-1] == 0 && c.recentL[len(c.recentL)-2] == 0 &&
+		c.recentQ[len(c.recentQ)-1] > 0 && c.recentQ[len(c.recentQ)-2] > 0
+	c.mu.Unlock()
+	if quiet && !final {
+		c.ensureLogging(ctx)
+	}
+	// Prune once an hour.
+	if now.Minute() == 0 || final {
+		c.prune(ctx)
+	}
+}
+
+// ensureLogging makes sure named logs responses (at most every 5 minutes).
+func (c *Collector) ensureLogging(ctx context.Context) {
+	if !c.opts.QueryLog || c.opts.EnsureLogging == nil || c.now().Sub(c.ensureAt) < 5*time.Minute {
+		return
+	}
+	c.ensureAt = c.now()
+	changed, err := c.opts.EnsureLogging(ctx)
+	switch {
+	case err != nil:
+		c.log.Warn("statistics: could not turn on named's response logging (rndc responselog on)", "err", err)
+	case changed:
+		c.log.Info("statistics: turned on named's response logging (rndc responselog on)")
+	}
+}
+
+// noteRecent records the last ten minutes of query counts and log lines
+// for the health warnings.
+func (c *Collector) noteRecent(queries int64) {
+	c.recentQ = append(c.recentQ, queries)
+	c.recentL = append(c.recentL, c.linesMin)
+	c.linesMin = 0
+	if len(c.recentQ) > 10 {
+		c.recentQ, c.recentL = c.recentQ[1:], c.recentL[1:]
+	}
+}
+
+func (c *Collector) updateWarnings() {
+	w := []string{}
+	if c.status.PollError != "" {
+		w = append(w, c.status.PollError)
+	}
+	if c.opts.QueryLog {
+		var q, l int64
+		for i := range c.recentQ {
+			q += c.recentQ[i]
+			l += c.recentL[i]
+		}
+		if len(c.recentQ) >= 5 && q > 0 && l == 0 {
+			w = append(w, fmt.Sprintf("named answered %d queries in the last %d minutes, but Jarl read no response log lines from %s: check that /var/log/named exists and is writable by named, and that responselog is not turned off", q, len(c.recentQ), c.opts.LogFile))
+		}
+		if c.status.LogError != "" {
+			w = append(w, "reading the response log failed: "+c.status.LogError)
+		}
+	}
+	c.status.Warnings = w
+}
+
+// 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{k, 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)
+		}
+	}
+}
+
+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/collector_test.go

@@ -0,0 +1,235 @@
+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{"nas.adbraun.lol", 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{"nas.adbraun.lol", 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{"nas.adbraun.lol", 4}) {
+		t.Errorf("hour 10 after late line = %v", h)
+	}
+	if d, _ := store.TopsAt(ctx, ResDay, day, KindName); d[0] != (Entry{"nas.adbraun.lol", 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{"printer.adbraun.lol", 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{"adbraun.lol", 7}) || b[1] != (Entry{OtherZone, 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)
+		}
+	}
+}

internal/stats/parse.go

@@ -0,0 +1,60 @@
+package stats
+
+import (
+	"net/netip"
+	"regexp"
+	"strings"
+	"time"
+)
+
+// Event is one response named logged.
+type Event struct {
+	Time   time.Time
+	Client netip.Addr
+	View   string // "" without views
+	Name   string // lowercase, without the final dot ("." for the root)
+	Type   string // e.g. A, AAAA, PTR
+	RCode  string // e.g. NOERROR, NXDOMAIN
+}
+
+// responseRe matches named's response log lines (BIND 9.20, print-time):
+//
+//	07-Oct-2026 19:46:22.195 client @0x7135... 10.10.0.37#51709 (nas.adbraun.lol): view internal: response: nas.adbraun.lol IN A NOERROR 1 0 1 +E(0)K (10.10.0.2)
+//
+// The "view NAME: " part is missing when no views are configured.
+var responseRe = regexp.MustCompile(`^(\d{2}-[A-Za-z]{3}-\d{4} \d{2}:\d{2}:\d{2}\.\d{3}) client @0x[0-9a-fA-F]+ (\S+)#\d+ \(.*?\): (?:view (.+?): )?response: (\S+) (\S+) (\S+) ([A-Z0-9]+) `)
+
+const logTimeLayout = "02-Jan-2006 15:04:05.000"
+
+// ParseResponse parses a response log line. ok is false for other lines
+// (named also logs to the channel when it starts and rotates files).
+func ParseResponse(line string, loc *time.Location) (ev Event, ok bool) {
+	m := responseRe.FindStringSubmatch(line)
+	if m == nil {
+		return ev, false
+	}
+	t, err := time.ParseInLocation(logTimeLayout, m[1], loc)
+	if err != nil {
+		return ev, false
+	}
+	client, err := netip.ParseAddr(m[2])
+	if err != nil {
+		return ev, false
+	}
+	return Event{
+		Time:   t,
+		Client: client.Unmap(),
+		View:   m[3],
+		Name:   normalizeName(m[4]),
+		Type:   m[6],
+		RCode:  m[7],
+	}, true
+}
+
+func normalizeName(n string) string {
+	n = strings.ToLower(n)
+	if n == "." {
+		return n
+	}
+	return strings.TrimSuffix(n, ".")
+}

internal/stats/parse_test.go

@@ -0,0 +1,44 @@
+package stats
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParseResponse(t *testing.T) {
+	loc := time.FixedZone("test", 3600)
+	for _, tc := range []struct {
+		line                        string
+		ok                          bool
+		client, view, name, typ, rc string
+	}{
+		// Real lines from named 9.20.24 (with and without views).
+		{`07-Oct-2026 19:46:22.184 client @0x71358b8ff000 127.0.0.1#53937 (www.example.test): view internal: response: www.example.test IN A NOERROR 1 0 1 +E(0)K (127.0.0.1)`,
+			true, "127.0.0.1", "internal", "www.example.test", "A", "NOERROR"},
+		{`07-Oct-2026 19:46:22.195 client @0x71358b8ff000 10.10.0.37#51709 (nope.example.test): response: Nope.Example.Test IN AAAA NXDOMAIN 0 1 1 +E(0)K (127.0.0.1)`,
+			true, "10.10.0.37", "", "nope.example.test", "AAAA", "NXDOMAIN"},
+		{`07-Oct-2026 19:46:22.217 client @0x71358b8ff000 2001:db8::7#55006 (other.zone): view external net: response: other.zone. IN MX REFUSED 0 0 1 +E(0)K (2001:db8::1)`,
+			true, "2001:db8::7", "external net", "other.zone", "MX", "REFUSED"},
+		{`07-Oct-2026 19:46:22.217 client @0x71358b8ff000 ::ffff:10.0.0.1#55006 (.): response: . IN NS NOERROR 13 0 27 + (10.0.0.2)`,
+			true, "10.0.0.1", "", ".", "NS", "NOERROR"},
+		// Query lines and other messages are not responses.
+		{`07-Oct-2026 19:46:22.184 client @0x71358b8ff000 127.0.0.1#53937 (www.example.test): view internal: query: www.example.test IN A +E(0)K (127.0.0.1)`, false, "", "", "", "", ""},
+		{`07-Oct-2026 19:46:20.000 logging channel 'jarl_responses' file '/var/log/named/jarl-responses.log': rolled`, false, "", "", "", "", ""},
+		{``, false, "", "", "", "", ""},
+	} {
+		ev, ok := ParseResponse(tc.line, loc)
+		if ok != tc.ok {
+			t.Errorf("%q: ok = %v", tc.line, ok)
+			continue
+		}
+		if !ok {
+			continue
+		}
+		if ev.Client.String() != tc.client || ev.View != tc.view || ev.Name != tc.name || ev.Type != tc.typ || ev.RCode != tc.rc {
+			t.Errorf("%q: got %+v", tc.line, ev)
+		}
+		if want := time.Date(2026, 10, 7, 19, 46, 22, 0, loc); ev.Time.Truncate(time.Second) != want {
+			t.Errorf("%q: time %v, want %v", tc.line, ev.Time, want)
+		}
+	}
+}

internal/stats/poll.go

@@ -0,0 +1,154 @@
+package stats
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// Snapshot is named's cumulative counters at one moment, flattened to the
+// keys Jarl stores (see Counters in docs/stats.md).
+type Snapshot struct {
+	BootTime string
+	Values   map[string]int64
+}
+
+// serverJSON is the part of /json/v1/server Jarl uses.
+type serverJSON struct {
+	BootTime string           `json:"boot-time"`
+	Opcodes  map[string]int64 `json:"opcodes"`
+	RCodes   map[string]int64 `json:"rcodes"`
+	QTypes   map[string]int64 `json:"qtypes"`
+	NSStats  map[string]int64 `json:"nsstats"`
+	Views    map[string]struct {
+		Resolver struct {
+			Stats      map[string]int64 `json:"stats"`
+			CacheStats map[string]int64 `json:"cachestats"`
+		} `json:"resolver"`
+	} `json:"views"`
+}
+
+// nsstatsKeys maps named's server counters to Jarl's counter names.
+var nsstatsKeys = map[string]string{
+	"QryAuthAns":   "answer:authoritative",
+	"QryNoauthAns": "answer:recursive",
+	"QryRecursion": "recursion",
+	"QryDropped":   "dropped",
+	"RateDropped":  "ratelimit:dropped",
+	"RateSlipped":  "ratelimit:slipped",
+	"QryTCP":       "transport:tcp",
+	"QryUDP":       "transport:udp",
+	"QryDuplicate": "duplicate",
+}
+
+// rttKeys maps named's upstream round-trip buckets to readable names.
+var rttKeys = map[string]string{
+	"QryRTT10":    "rtt:<10ms",
+	"QryRTT100":   "rtt:10-100ms",
+	"QryRTT500":   "rtt:100-500ms",
+	"QryRTT800":   "rtt:500-800ms",
+	"QryRTT1600":  "rtt:800-1600ms",
+	"QryRTT1600+": "rtt:>1600ms",
+}
+
+var dnssecKeys = map[string]string{"ValOk": "dnssec:valid", "ValNegOk": "dnssec:valid-negative", "ValFail": "dnssec:failed"}
+
+// Flatten turns the statistics channel's JSON into a snapshot.
+func Flatten(body []byte) (*Snapshot, error) {
+	var j serverJSON
+	if err := json.Unmarshal(body, &j); err != nil {
+		return nil, fmt.Errorf("unexpected statistics data: %w", err)
+	}
+	v := map[string]int64{"queries": j.Opcodes["QUERY"]}
+	for k, n := range j.RCodes {
+		if n != 0 && !strings.HasPrefix(k, "RESERVED") && (k[0] < '0' || k[0] > '9') {
+			v["rcode:"+k] = n
+		}
+	}
+	for k, n := range j.QTypes {
+		v["qtype:"+k] = n
+	}
+	for k, name := range nsstatsKeys {
+		if n := j.NSStats[k]; n != 0 {
+			v[name] = n
+		}
+	}
+	for _, view := range j.Views {
+		for k, name := range rttKeys {
+			v[name] += view.Resolver.Stats[k]
+		}
+		for k, name := range dnssecKeys {
+			v[name] += view.Resolver.Stats[k]
+		}
+		v["cache:hits"] += view.Resolver.CacheStats["QueryHits"]
+		v["cache:misses"] += view.Resolver.CacheStats["QueryMisses"]
+	}
+	for k, n := range v {
+		if n == 0 && k != "queries" {
+			delete(v, k)
+		}
+	}
+	return &Snapshot{BootTime: j.BootTime, Values: v}, nil
+}
+
+// Delta returns the change from prev to cur. When named restarted (its
+// boot time changed or a counter went down), the counters started again
+// from zero, so the current values are the change.
+func Delta(prev, cur *Snapshot) map[string]int64 {
+	out := map[string]int64{}
+	restarted := prev.BootTime != cur.BootTime
+	if !restarted {
+		for k, n := range cur.Values {
+			if n < prev.Values[k] {
+				restarted = true
+				break
+			}
+		}
+	}
+	for k, n := range cur.Values {
+		d := n
+		if !restarted {
+			d = n - prev.Values[k]
+		}
+		if d != 0 || k == "queries" {
+			out[k] = d
+		}
+	}
+	return out
+}
+
+// Poller reads named's statistics channel.
+type Poller struct {
+	URL    string // e.g. http://127.0.0.1:8058/json/v1/server
+	Client *http.Client
+}
+
+// NewPoller returns a poller for a statistics channel address.
+func NewPoller(addr string) *Poller {
+	return &Poller{URL: "http://" + addr + "/json/v1/server", Client: &http.Client{Timeout: 10 * time.Second}}
+}
+
+// Poll fetches a snapshot.
+func (p *Poller) Poll(ctx context.Context) (*Snapshot, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.URL, nil)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := p.Client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("named's statistics channel is not reachable (%v); is the configuration with statistics applied and named running?", err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("named's statistics channel answered %s", resp.Status)
+	}
+	return Flatten(body)
+}

internal/stats/poll_test.go

@@ -0,0 +1,55 @@
+package stats
+
+import "testing"
+
+// A trimmed /json/v1/server answer from named 9.20.24.
+const sampleJSON = `{"json-stats-version":"1.8","boot-time":"2026-10-07T19:46:51.322Z",
+ "opcodes":{"QUERY":10,"IQUERY":0,"NOTIFY":0},
+ "rcodes":{"NOERROR":7,"NXDOMAIN":2,"SERVFAIL":1,"RESERVED11":0,"17":0},
+ "qtypes":{"A":6,"AAAA":3,"MX":1},
+ "nsstats":{"Requestv4":10,"QryAuthAns":4,"QryNoauthAns":6,"QryRecursion":2,"QryUDP":9,"QryTCP":1},
+ "views":{"internal":{"resolver":{"stats":{"QryRTT100":3,"QryRTT500":1,"ValOk":2},"cachestats":{"QueryHits":4,"QueryMisses":2,"CacheHits":345}}},
+          "_bind":{"resolver":{"stats":{},"cachestats":{}}}}}`
+
+func TestFlattenAndDelta(t *testing.T) {
+	s1, err := Flatten([]byte(sampleJSON))
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]int64{"queries": 10, "rcode:NOERROR": 7, "rcode:NXDOMAIN": 2, "rcode:SERVFAIL": 1, "qtype:A": 6, "qtype:AAAA": 3, "qtype:MX": 1,
+		"answer:authoritative": 4, "answer:recursive": 6, "recursion": 2, "transport:udp": 9, "transport:tcp": 1,
+		"rtt:10-100ms": 3, "rtt:100-500ms": 1, "dnssec:valid": 2, "cache:hits": 4, "cache:misses": 2}
+	for k, v := range want {
+		if s1.Values[k] != v {
+			t.Errorf("%s = %d, want %d", k, s1.Values[k], v)
+		}
+	}
+	if len(s1.Values) != len(want) {
+		t.Errorf("unexpected keys: %v", s1.Values)
+	}
+	// Normal change.
+	s2 := &Snapshot{BootTime: s1.BootTime, Values: map[string]int64{}}
+	for k, v := range s1.Values {
+		s2.Values[k] = v
+	}
+	s2.Values["queries"] += 5
+	s2.Values["qtype:A"] += 5
+	d := Delta(s1, s2)
+	if d["queries"] != 5 || d["qtype:A"] != 5 || d["qtype:AAAA"] != 0 || len(d) != 2 {
+		t.Errorf("delta = %v", d)
+	}
+	// Quiet minute: queries is still recorded (as 0) so coverage is visible.
+	if d := Delta(s2, s2); len(d) != 1 || d["queries"] != 0 {
+		t.Errorf("quiet delta = %v", d)
+	}
+	// named restarted: counters start again from zero.
+	s3 := &Snapshot{BootTime: "2026-10-07T20:00:00.000Z", Values: map[string]int64{"queries": 3, "qtype:A": 3}}
+	if d := Delta(s2, s3); d["queries"] != 3 || d["qtype:A"] != 3 {
+		t.Errorf("restart delta = %v", d)
+	}
+	// A counter going down also means a restart.
+	s4 := &Snapshot{BootTime: s2.BootTime, Values: map[string]int64{"queries": 2}}
+	if d := Delta(s2, s4); d["queries"] != 2 {
+		t.Errorf("reset delta = %v", d)
+	}
+}

internal/stats/query.go

@@ -0,0 +1,422 @@
+package stats
+
+import (
+	"cmp"
+	"context"
+	"fmt"
+	"slices"
+	"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: time.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 in [from, to).
+func (s *Store) tops(ctx context.Context, res, kind string, from, to int64, limit int) ([]Entry, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT key, SUM(count) AS n FROM tops WHERE res = ? AND kind = ? AND ts >= ? AND ts < ?
+		GROUP BY key ORDER BY n DESC, key LIMIT ?`, res, kind, from, to, limit)
+	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{strings.TrimPrefix(k, prefix), 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
+}
+
+// Summary is an overview of a time range.
+type Summary struct {
+	From       time.Time `json:"from"`
+	To         time.Time `json:"to"`
+	Resolution string    `json:"resolution"`
+	Queries    int64     `json:"queries"`
+	// AvgQPS is queries per second over the part of the range with data.
+	AvgQPS float64 `json:"avg_qps"`
+	Peak   *struct {
+		Time    time.Time `json:"time"`
+		Queries int64     `json:"queries"`
+		Per     string    `json:"per"` // minute, hour or day
+	} `json:"peak,omitempty"`
+	RCodes        []Entry            `json:"rcodes"`
+	QTypes        []Entry            `json:"qtypes"`
+	Views         []Entry            `json:"views"`
+	Zones         []Entry            `json:"zones"`
+	Answers       []Entry            `json:"answers"`   // authoritative, recursive
+	Transport     []Entry            `json:"transport"` // udp, tcp
+	CacheHitRatio *float64           `json:"cache_hit_ratio,omitempty"`
+	NXDomainRatio *float64           `json:"nxdomain_ratio,omitempty"`
+	Tops          map[string][]Entry `json:"tops,omitempty"`
+}
+
+// Summary summarizes the last since.
+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))
+	to := now.Unix() + 1
+	all, err := q.Store.sums(ctx, res, from, to, "")
+	if err != nil {
+		return nil, err
+	}
+	s := &Summary{From: time.Unix(from, 0), To: now, Resolution: res, Queries: all["queries"],
+		RCodes: sorted(all, "rcode:"), QTypes: sorted(all, "qtype:"), Views: sorted(all, "view:"), Zones: sorted(all, "zone:"),
+		Answers: sorted(all, "answer:"), Transport: sorted(all, "transport:")}
+	if h, m := all["cache:hits"], all["cache:misses"]; h+m > 0 {
+		r := float64(h) / float64(h+m)
+		s.CacheHitRatio = &r
+	}
+	if s.Queries > 0 {
+		r := float64(all["rcode:NXDOMAIN"]) / float64(s.Queries)
+		s.NXDomainRatio = &r
+	}
+	per, err := q.Store.series(ctx, res, from, to, "queries")
+	if err != nil {
+		return nil, err
+	}
+	if len(per) > 0 {
+		first, peakTS := int64(1<<62), int64(0)
+		var peak int64 = -1
+		for ts, v := range per {
+			first = min(first, ts)
+			if v > peak || v == peak && ts < peakTS {
+				peak, peakTS = v, ts
+			}
+		}
+		if secs := float64(to - first); secs > 0 {
+			s.AvgQPS = float64(s.Queries) / secs
+		}
+		s.Peak = &struct {
+			Time    time.Time `json:"time"`
+			Queries int64     `json:"queries"`
+			Per     string    `json:"per"`
+		}{time.Unix(peakTS, 0), peak, map[string]string{ResMinute: "minute", ResHour: "hour", ResDay: "day"}[res]}
+	}
+	if topLimit > 0 {
+		s.Tops = map[string][]Entry{}
+		for _, k := range []string{KindName, KindClient, KindNXDomain} {
+			es, err := q.Top(ctx, k, since, topLimit)
+			if err != nil {
+				return nil, err
+			}
+			s.Tops[k] = es
+		}
+	}
+	return s, nil
+}
+
+// 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) {
+		return nil, fmt.Errorf("unknown list %q (use %s)", kind, strings.Join(Kinds, ", "))
+	}
+	now := q.Now()
+	res := q.topsResolution(since)
+	return q.Store.tops(ctx, res, kind, q.bucketStart(res, now.Add(-since)), now.Unix()+1, limit)
+}
+
+// Breakdown returns the totals of a counter group (qtype, rcode, view,
+// zone, answer, transport, rtt, dnssec, cache, ratelimit) over the last since.
+func (q *Querier) Breakdown(ctx context.Context, group string, since time.Duration) ([]Entry, error) {
+	if !slices.Contains(Groups, group) {
+		return nil, fmt.Errorf("unknown group %q (use %s)", group, strings.Join(Groups, ", "))
+	}
+	now := q.Now()
+	res := q.Resolution(since)
+	m, err := q.Store.sums(ctx, res, q.bucketStart(res, now.Add(-since)), now.Unix()+1, group+":")
+	if err != nil {
+		return nil, err
+	}
+	return sorted(m, group+":"), nil
+}
+
+// Groups are the counter groups Breakdown accepts.
+var Groups = []string{"qtype", "rcode", "view", "zone", "answer", "transport", "rtt", "dnssec", "cache", "ratelimit"}
+
+// 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.
+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{}}
+	for _, k := range expanded {
+		raw, err := q.Store.series(ctx, st.res, from, to, k)
+		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
+			}
+		}
+		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 "queries") 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 = "queries"
+	}
+	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
+}

internal/stats/store.go

@@ -0,0 +1,254 @@
+// Package stats collects query statistics from named: counters from its
+// statistics channel and per-query data from its response log. It stores
+// aggregates (never individual queries) in a SQLite database of its own,
+// separate from Jarl's configuration database, so it can be deleted at any
+// time without losing anything important.
+package stats
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+
+	_ "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
+}
+
+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);`,
+}
+
+// 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}
+	if err := s.migrate(); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("statistics database %s: %w", path, err)
+	}
+	return s, nil
+}
+
+// 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 Jarl; upgrade Jarl 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"`
+}
+
+// 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()
+}

internal/stats/tail.go

@@ -0,0 +1,159 @@
+package stats
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"io/fs"
+	"os"
+	"syscall"
+)
+
+// Tailer follows a log file that named rotates (renaming it to FILE.0,
+// FILE.1, ... and starting a new FILE). It remembers its position as
+// (inode, offset) so reading resumes after a restart.
+type Tailer struct {
+	Path string
+
+	f       *os.File
+	ino     uint64
+	offset  int64
+	partial []byte
+	// draining is set while finishing a rotated file before switching
+	// to the current one.
+	draining bool
+}
+
+// Position is where the tailer is, for persisting.
+type Position struct {
+	Inode  uint64 `json:"inode"`
+	Offset int64  `json:"offset"`
+}
+
+func inode(fi fs.FileInfo) uint64 {
+	if st, ok := fi.Sys().(*syscall.Stat_t); ok {
+		return st.Ino
+	}
+	return 0
+}
+
+// Position returns the current position (of complete lines).
+func (t *Tailer) Position() Position {
+	return Position{Inode: t.ino, Offset: t.offset - int64(len(t.partial))}
+}
+
+// Resume continues from a saved position: in the current file, or in the
+// newest rotated file if named rotated it in the meantime. Without a
+// usable position, reading starts at the beginning of the current file.
+func (t *Tailer) Resume(p Position) {
+	if p.Inode == 0 {
+		return
+	}
+	for _, path := range []string{t.Path, t.Path + ".0"} {
+		fi, err := os.Stat(path)
+		if err != nil || inode(fi) != p.Inode || fi.Size() < p.Offset {
+			continue
+		}
+		f, err := os.Open(path)
+		if err != nil {
+			return
+		}
+		if _, err := f.Seek(p.Offset, io.SeekStart); err != nil {
+			f.Close()
+			return
+		}
+		t.f, t.ino, t.offset, t.draining = f, p.Inode, p.Offset, path != t.Path
+		return
+	}
+}
+
+// open opens the current file from the start. A missing file is not an
+// error: named creates it when it logs the first response.
+func (t *Tailer) open() (bool, error) {
+	f, err := os.Open(t.Path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	fi, err := f.Stat()
+	if err != nil {
+		f.Close()
+		return false, err
+	}
+	t.f, t.ino, t.offset, t.partial, t.draining = f, inode(fi), 0, nil, false
+	return true, nil
+}
+
+// Read calls fn for every complete new line and returns how many lines it
+// read. Call it periodically.
+func (t *Tailer) Read(fn func(line string)) (int, error) {
+	if t.f == nil {
+		ok, err := t.open()
+		if !ok || err != nil {
+			return 0, err
+		}
+	}
+	n := 0
+	buf := make([]byte, 64<<10)
+	for {
+		k, err := t.f.Read(buf)
+		if k > 0 {
+			t.offset += int64(k)
+			data := append(t.partial, buf[:k]...)
+			for {
+				i := bytes.IndexByte(data, '\n')
+				if i < 0 {
+					break
+				}
+				fn(string(data[:i]))
+				n++
+				data = data[i+1:]
+			}
+			t.partial = append([]byte(nil), data...)
+			if len(t.partial) > 1<<20 { // not a log line; drop it
+				t.partial = nil
+			}
+		}
+		if err == io.EOF || k == 0 {
+			break
+		}
+		if err != nil {
+			return n, err
+		}
+	}
+	// At the end of the file: has named rotated or truncated it?
+	fi, err := os.Stat(t.Path)
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+		return n, nil // rotated and not recreated yet
+	case err != nil:
+		return n, err
+	case t.draining || inode(fi) != t.ino:
+		t.f.Close()
+		t.f = nil
+		if _, err := t.open(); err != nil {
+			return n, err
+		}
+		if t.f != nil {
+			m, err := t.Read(fn)
+			return n + m, err
+		}
+	case fi.Size() < t.offset:
+		// Truncated (e.g. by hand): start again from the top.
+		if _, err := t.f.Seek(0, io.SeekStart); err != nil {
+			return n, err
+		}
+		t.offset, t.partial = 0, nil
+	}
+	return n, nil
+}
+
+// Close closes the file.
+func (t *Tailer) Close() {
+	if t.f != nil {
+		t.f.Close()
+		t.f = nil
+	}
+}

internal/stats/tail_test.go

@@ -0,0 +1,77 @@
+package stats
+
+import (
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+)
+
+func TestTailer(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "r.log")
+	var got []string
+	collect := func(l string) { got = append(got, l) }
+	appendTo := func(p, s string) {
+		f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+		if err != nil {
+			t.Fatal(err)
+		}
+		f.WriteString(s)
+		f.Close()
+	}
+	tl := &Tailer{Path: path}
+	// The file doesn't exist yet.
+	if n, err := tl.Read(collect); n != 0 || err != nil {
+		t.Fatalf("missing file: %d, %v", n, err)
+	}
+	appendTo(path, "one\ntwo\nthr")
+	tl.Read(collect)
+	appendTo(path, "ee\n")
+	tl.Read(collect)
+	if !reflect.DeepEqual(got, []string{"one", "two", "three"}) {
+		t.Fatalf("got %q", got)
+	}
+	// named rotates: renames the file and starts a new one. Lines written
+	// to the old file before the rename are not lost.
+	appendTo(path, "four\n")
+	os.Rename(path, path+".0")
+	appendTo(path, "five\n")
+	tl.Read(collect)
+	if !reflect.DeepEqual(got[3:], []string{"four", "five"}) {
+		t.Fatalf("after rotation got %q", got)
+	}
+	// Restart: resume from the saved position.
+	pos := tl.Position()
+	tl.Close()
+	appendTo(path, "six\n")
+	tl2 := &Tailer{Path: path}
+	tl2.Resume(pos)
+	got = nil
+	tl2.Read(collect)
+	if !reflect.DeepEqual(got, []string{"six"}) {
+		t.Fatalf("after resume got %q", got)
+	}
+	// Rotation while jarld was down: finish the rotated file first.
+	pos = tl2.Position()
+	tl2.Close()
+	appendTo(path, "seven\n")
+	os.Rename(path, path+".0")
+	appendTo(path, "eight\n")
+	tl3 := &Tailer{Path: path}
+	tl3.Resume(pos)
+	got = nil
+	tl3.Read(collect)
+	if !reflect.DeepEqual(got, []string{"seven", "eight"}) {
+		t.Fatalf("after offline rotation got %q", got)
+	}
+	// Truncation: start over.
+	os.WriteFile(path, []byte("x\n"), 0o644)
+	got = nil
+	tl3.Read(collect)
+	tl3.Read(collect)
+	if !reflect.DeepEqual(got, []string{"x"}) {
+		t.Fatalf("after truncation got %q", got)
+	}
+	tl3.Close()
+}