package stats import ( "cmp" "context" "encoding/json" "fmt" "log/slog" "net/netip" "slices" "strconv" "strings" "sync" "time" "jarl/internal/blocking" ) // 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 KindBlocked = "blocked" // names blocked (docs/blocking.md) // KindBlockedClient: clients with the most blocked queries. KindBlockedClient = "blocked_client" // KindRateLimited: client networks whose responses rate limiting // limited (or would have, in log-only mode); KindRateLimitedName: the // names and types they asked for (docs/exposure.md). KindRateLimited = "ratelimited" KindRateLimitedName = "ratelimited_name" ) // ViewKey is the key of a counter for one view: per-view copies of the // counters that come from named's response log (docs/stats.md). func ViewKey(view, key string) string { return "@" + view + "/" + key } // ViewKind is the kind of a top list for one view. func ViewKind(kind, view string) string { return kind + "@" + view } // Kinds lists the top-list kinds. var Kinds = []string{KindName, KindClient, KindNXDomain, KindServFail, KindRefused, KindBlocked, KindBlockedClient, KindRateLimited, KindRateLimitedName} // 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) // MainConfig is named's main configuration file; it tells named's // process apart from other named processes (for its memory and CPU). MainConfig string // 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 // rewritten holds queries named blocked (client#port|name) until their // response line arrives. rewritten map[string]bool // viewsSince is set once per-view counters exist (meta views_since). viewsSince bool viewsSinceAt int64 // proc is the last sample of named's process, server the latest server // statistics (server.go). proc *namedProc server *Server // totals are the counters added up since jarld started (gauges left // out), for Prometheus (Totals). totals map[string]int64 } // 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{}, totals: 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, "views_since"); v != "" { c.viewsSince = true } 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 { 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)) curKinds, _ := c.store.KindsAt(ctx, ResHour, cur) for _, k := range curKinds { 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. snap, err := c.poller.Poll(ctx) if err == nil { c.prev = snap t := c.now() c.status.LastPoll = &t } else { c.status.PollError = err.Error() } c.mu.Lock() c.sampleServer(-1, c.now(), snap) c.mu.Unlock() 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 if !strings.HasPrefix(key, "g:") && !strings.HasPrefix(key, "gn:") && !strings.HasPrefix(key, "@") { c.totals[key] += n } } // Totals returns every counter added up since the collector started (the // keys of docs/stats.md, without gauges). They only grow, so they suit // Prometheus counters; named's counters move once a minute, when they are // polled. func (c *Collector) Totals() map[string]int64 { c.mu.Lock() defer c.mu.Unlock() out := make(map[string]int64, len(c.totals)) for k, v := range c.totals { out[k] = v } return out } // 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 { if rw, ok := ParseRewrite(line, c.opts.Location); ok { c.handleRewrite(rw) } else if rl, ok := ParseRateLimit(line, c.opts.Location); ok { c.handleRateLimit(rl) } return } blocked := c.rewritten[ev.Client.String()+"#"+ev.Port+"|"+ev.Name] if blocked { delete(c.rewritten, ev.Client.String()+"#"+ev.Port+"|"+ev.Name) } 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" } zone := c.zoneOf(ev.Name, ev.View) c.count(minute, "responses", 1) c.count(minute, "view:"+view, 1) c.count(minute, "zone:"+zone, 1) // The same, per view, for the statistics of chosen views. c.countView(minute, view, "responses", 1) c.countView(minute, view, "zone:"+zone, 1) c.countView(minute, view, "rcode:"+ev.RCode, 1) c.countView(minute, view, "qtype:"+ev.Type, 1) client := clientKey(ev.Client, c.opts.Anonymize) hours := c.hours if hour < c.liveHour { hours = c.late } hours[hour] = c.hourMaps(hours, hour) add := func(kind, key string) { hours[hour][kind][key]++ m := c.topsMap(hours, hour, ViewKind(kind, view)) m[key]++ if len(m) > maxKeys { shrink(m) } } add(KindName, ev.Name) add(KindClient, client) switch { case blocked: // Counted by handleRewrite; not a name that doesn't exist. case ev.RCode == "NXDOMAIN": add(KindNXDomain, ev.Name) case ev.RCode == "SERVFAIL": add(KindServFail, ev.Name) case ev.RCode == "REFUSED": add(KindRefused, client) } for _, k := range Kinds { if m := hours[hour][k]; len(m) > maxKeys { shrink(m) } } } // countView counts a per-view copy of a counter. func (c *Collector) countView(minute int64, view, key string, n int64) { if !c.viewsSince { c.viewsSince = true c.viewsSinceAt = minute } c.count(minute, ViewKey(view, key), n) } // handleRewrite counts a query that blocking answered: "blocked", and // "blocklist:NAME" for the list (or "rules") that matched. func (c *Collector) handleRewrite(rw Rewrite) { if rw.Action == "PASSTHRU" { return // an allow rule: answered normally } for _, p := range c.opts.Exclude { if p.Contains(rw.Client) { return } } if c.rewritten == nil || len(c.rewritten) > 10_000 { // Rewrites whose response line never came (named only logs // responses while response logging is on) don't pile up. c.rewritten = map[string]bool{} } c.rewritten[rw.Client.String()+"#"+rw.Port+"|"+rw.Name] = true minute, hour := c.minuteOf(rw.Time), c.hourOf(rw.Time) view := rw.View if view == "" { view = "_default" } c.count(minute, "blocked", 1) c.count(minute, "blocklist:"+policySource(rw.Via), 1) c.countView(minute, view, "blocked", 1) c.countView(minute, view, "blocklist:"+policySource(rw.Via), 1) hours := c.hours if hour < c.liveHour { hours = c.late } hours[hour] = c.hourMaps(hours, hour) client := clientKey(rw.Client, c.opts.Anonymize) hours[hour][KindBlocked][rw.Name]++ hours[hour][KindBlockedClient][client]++ c.topsMap(hours, hour, ViewKind(KindBlocked, view))[rw.Name]++ c.topsMap(hours, hour, ViewKind(KindBlockedClient, view))[client]++ } // handleRateLimit counts a response rate limiting limited, or would have: // "rrl:drop", "rrl:slip", "rrl:would-drop" or "rrl:would-slip", and the // client network and name in the hour's top lists. func (c *Collector) handleRateLimit(rl RateLimit) { for _, p := range c.opts.Exclude { if p.Contains(rl.Client) { return } } key := "rrl:" + rl.Action if rl.Would { key = "rrl:would-" + rl.Action } minute, hour := c.minuteOf(rl.Time), c.hourOf(rl.Time) view := rl.View if view == "" { view = "_default" } c.count(minute, key, 1) c.countView(minute, view, key, 1) hours := c.hours if hour < c.liveHour { hours = c.late } hours[hour] = c.hourMaps(hours, hour) hours[hour][KindRateLimited][rl.Prefix]++ hours[hour][KindRateLimitedName][rl.Name+" "+rl.Type]++ c.topsMap(hours, hour, ViewKind(KindRateLimited, view))[rl.Prefix]++ c.topsMap(hours, hour, ViewKind(KindRateLimitedName, view))[rl.Name+" "+rl.Type]++ } // policySource names what a rewrite came from: a blocklist's name, "rules" // for the rules entered in Jarl, or the policy zone for others. func policySource(via string) string { via = strings.TrimSuffix(via, ".") if strings.HasSuffix(via, "."+blocking.RulesZone) { return "rules" } if i := strings.LastIndex(via, ".blocklist.jarl.invalid"); i > 0 && i+len(".blocklist.jarl.invalid") == len(via) { rest := via[:i] return rest[strings.LastIndex(rest, ".")+1:] } return via } // 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 } if !final { // Gauges are sampled once a minute, for the minute that ended. c.sampleServer(cur-60, now, 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 c.viewsSinceAt != 0 { b.SetMeta("views_since", strconv.FormatInt(c.viewsSinceAt, 10)) c.viewsSinceAt = 0 } 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{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) } } } 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() }