housecarl/internal/stats/query.go

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 and NAT objects, by packets; BuiltIns
	// the busiest built-ins and policies, kept apart because the
	// established built-in counts most of the traffic.
	Rules    []Entry `json:"rules"`
	BuiltIns []Entry `json:"builtins"`
	// 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, s.BuiltIns = []Entry{}, []Entry{}
	for _, e := range sorted(all, "hits:") {
		if strings.HasPrefix(e.Key, "b-") || strings.HasPrefix(e.Key, "p-") {
			if len(s.BuiltIns) < topLimit {
				s.BuiltIns = append(s.BuiltIns, e)
			}
		} else if len(s.Rules) < topLimit {
			s.Rules = append(s.Rules, e)
		}
	}
	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()
}