jarl/internal/stats/explore.go

package stats

import (
	"context"
	"fmt"
	"strings"
	"time"
)

// Exploring the query log (docs/stats.md#exploring): every answer of the
// last days (stats.retention.queries), filtered by any mix of client, name,
// view, type, result and blocking, broken down by each of them.

// ExploreFilter narrows the query log. Empty fields don't filter. Each
// field may hold several values, comma separated: any of them; a value
// written !VALUE excludes it. A name *.example.com is it and the names
// below it.
type ExploreFilter struct {
	Client string // a client key, as the top lists show it
	Name   string // exactly this name
	Domain string // this name or any below it
	View   string
	Type   string
	RCode  string
	// Blocked is "yes" (blocked by anything), "no", or a blocklist's name
	// ("rules" for your own rules).
	Blocked string
	Since   time.Duration // how far back (default 24h)
}

// Dimensions are what the query log breaks down by, in the order shown.
var Dimensions = []string{"client", "name", "view", "type", "rcode", "blocked"}

// Exploration is what the query log says for a filter.
type Exploration struct {
	From  time.Time `json:"from"`
	To    time.Time `json:"to"`
	Total int64     `json:"total"`
	// Blocked counts the blocked answers among them.
	Blocked int64 `json:"blocked"`
	// Series are the answers (and blocked ones) per Step seconds.
	Step   int64          `json:"step"`
	Series []ExplorePoint `json:"series"`
	// By are the top values of each dimension the filter leaves open.
	By map[string][]Entry `json:"by"`
	// Recent are the latest answers.
	Recent []QueryRow `json:"recent"`
	// Oldest is the first answer the query log still has.
	Oldest *time.Time `json:"oldest,omitempty"`
	// Retention is how long it keeps them, also in seconds.
	Retention        string `json:"retention"`
	RetentionSeconds int64  `json:"retention_seconds"`
}

// ExplorePoint is one step of the series.
type ExplorePoint struct {
	Time    int64 `json:"t"`
	Count   int64 `json:"count"`
	Blocked int64 `json:"blocked"`
}

// ErrNoQueryLog is returned when the query log is off.
var ErrNoQueryLog = fmt.Errorf("the query log is off: it needs stats.query_log and a stats.retention.queries above 0 in /etc/jarl/jarl.yaml (docs/stats.md#exploring)")

// values splits a filter field into the values it must be one of and the
// ones (written !VALUE) it must not be: "10.0.0.5, !10.0.0.9".
func values(field string, norm func(string) string) (in, out []string) {
	for _, v := range strings.Split(field, ",") {
		v = strings.TrimSpace(v)
		not := strings.HasPrefix(v, "!")
		v = norm(strings.TrimSpace(strings.TrimPrefix(v, "!")))
		switch {
		case v == "":
		case not:
			out = append(out, v)
		default:
			in = append(in, v)
		}
	}
	return in, out
}

// nameValues are the name filter's values: Name's, and Domain's as
// "*.DOMAIN" (the name and the names below it).
func (f ExploreFilter) nameValues() (in, out []string) {
	norm := func(v string) string { return strings.ToLower(strings.TrimSuffix(v, ".")) }
	in, out = values(f.Name, norm)
	din, dout := values(f.Domain, norm)
	for _, d := range din {
		in = append(in, "*."+strings.TrimPrefix(d, "*."))
	}
	for _, d := range dout {
		out = append(out, "*."+strings.TrimPrefix(d, "*."))
	}
	return in, out
}

// fieldValues returns a dimension's filter values.
func (f ExploreFilter) fieldValues(dim string) (in, out []string) {
	same := func(v string) string { return v }
	upper := strings.ToUpper
	switch dim {
	case "client":
		return values(f.Client, same)
	case "name":
		return f.nameValues()
	case "view":
		return values(f.View, same)
	case "type":
		return values(f.Type, upper)
	case "rcode":
		return values(f.RCode, upper)
	case "blocked":
		return values(f.Blocked, same)
	}
	return nil, nil
}

// cond is the SQL condition for one value of a dimension.
func cond(dim, v string) (string, []any) {
	switch {
	case dim == "name" && strings.HasPrefix(v, "*."):
		d := strings.TrimPrefix(v, "*.")
		return "(name = ? OR (length(name) > length(?) AND substr(name, length(name) - length(?)) = ?))", []any{d, d, d, "." + d}
	case dim == "blocked" && v == "yes":
		return "blocked != ''", nil
	case dim == "blocked" && v == "no":
		return "blocked = ''", nil
	}
	return dim + " = ?", []any{v}
}

// where builds the SQL condition and arguments for a filter: within a
// dimension any of its values, and none of its !values; the dimensions
// together.
func (q *Querier) where(f ExploreFilter, from int64) (string, []any) {
	conds := []string{"ts >= ?"}
	args := []any{from}
	for _, dim := range Dimensions {
		in, out := f.fieldValues(dim)
		if len(in) > 0 {
			var or []string
			for _, v := range in {
				c, a := cond(dim, v)
				or = append(or, c)
				args = append(args, a...)
			}
			conds = append(conds, "("+strings.Join(or, " OR ")+")")
		}
		for _, v := range out {
			c, a := cond(dim, v)
			conds = append(conds, "NOT "+c)
			args = append(args, a...)
		}
	}
	if len(q.Views) > 0 {
		marks := strings.TrimSuffix(strings.Repeat("?,", len(q.Views)), ",")
		conds = append(conds, "view IN ("+marks+")")
		for _, v := range q.Views {
			args = append(args, v)
		}
	}
	return strings.Join(conds, " AND "), args
}

// Explore answers a filter: totals, a series, the top values of each open
// dimension (topN each) and the latest answers (recent of them).
func (q *Querier) Explore(ctx context.Context, f ExploreFilter, topN, recent int) (*Exploration, error) {
	if !q.Opts.QueryLog || q.Opts.RetainQueries <= 0 {
		return nil, ErrNoQueryLog
	}
	now := q.Now()
	since := f.Since
	if since <= 0 {
		since = 24 * time.Hour
	}
	if since > q.Opts.RetainQueries {
		since = q.Opts.RetainQueries
	}
	from := now.Add(-since).Unix()
	out := &Exploration{From: time.Unix(from, 0), To: now, By: map[string][]Entry{}, Series: []ExplorePoint{}, Recent: []QueryRow{},
		Retention: humanDuration(q.Opts.RetainQueries), RetentionSeconds: int64(q.Opts.RetainQueries / time.Second)}
	cond, args := q.where(f, from)
	db := q.Store.db

	if err := db.QueryRowContext(ctx, `SELECT count(*), coalesce(sum(blocked != ''), 0) FROM queries WHERE `+cond, args...).Scan(&out.Total, &out.Blocked); err != nil {
		return nil, err
	}
	// The series: about 100 steps over the range, in whole minutes.
	step := int64(since.Seconds()) / 100
	step = max(60, step-step%60)
	out.Step = step
	rows, err := db.QueryContext(ctx, `SELECT ts - ts % ? AS t, count(*), sum(blocked != '') FROM queries WHERE `+cond+` GROUP BY t ORDER BY t`, append([]any{step}, args...)...)
	if err != nil {
		return nil, err
	}
	got := map[int64]ExplorePoint{}
	for rows.Next() {
		var p ExplorePoint
		if err := rows.Scan(&p.Time, &p.Count, &p.Blocked); err != nil {
			rows.Close()
			return nil, err
		}
		got[p.Time] = p
	}
	rows.Close()
	for t := from - from%step; t <= now.Unix(); t += step {
		p := got[t]
		p.Time = t
		out.Series = append(out.Series, p)
	}

	// A dimension is broken down unless the filter pins it to one value
	// (several: how they compare; !values leave the rest to see).
	open := map[string]bool{}
	for _, dim := range Dimensions {
		in, _ := f.fieldValues(dim)
		open[dim] = len(in) != 1 || (dim == "blocked" && in[0] == "yes") || (dim == "name" && strings.HasPrefix(in[0], "*."))
	}
	for _, dim := range Dimensions {
		if !open[dim] {
			continue
		}
		c := cond
		if dim == "blocked" {
			c += " AND blocked != ''"
		}
		rows, err := db.QueryContext(ctx, `SELECT `+dim+`, count(*) AS n FROM queries WHERE `+c+` GROUP BY `+dim+` ORDER BY n DESC, `+dim+` LIMIT ?`, append(args, topN)...)
		if err != nil {
			return nil, err
		}
		es := []Entry{}
		for rows.Next() {
			var e Entry
			if err := rows.Scan(&e.Key, &e.Count); err != nil {
				rows.Close()
				return nil, err
			}
			es = append(es, e)
		}
		rows.Close()
		out.By[dim] = es
	}

	rows, err = db.QueryContext(ctx, `SELECT ts, client, view, name, type, rcode, blocked FROM queries WHERE `+cond+` ORDER BY ts DESC, rowid DESC LIMIT ?`, append(args, recent)...)
	if err != nil {
		return nil, err
	}
	for rows.Next() {
		var r QueryRow
		if err := rows.Scan(&r.Time, &r.Client, &r.View, &r.Name, &r.Type, &r.RCode, &r.Blocked); err != nil {
			rows.Close()
			return nil, err
		}
		out.Recent = append(out.Recent, r)
	}
	rows.Close()

	var oldest *int64
	if err := db.QueryRowContext(ctx, `SELECT min(ts) FROM queries`).Scan(&oldest); err == nil && oldest != nil {
		t := time.Unix(*oldest, 0)
		out.Oldest = &t
	}
	return out, nil
}

// QueryLogSize returns how many answers the query log has.
func (s *Store) QueryLogSize(ctx context.Context) (int64, error) {
	var n int64
	err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM queries`).Scan(&n)
	return n, err
}