//go:build !js && !wasip1
package obsduck
import (
"context"
"database/sql"
"database/sql/driver"
"fmt"
"math"
"net/http"
"strconv"
"strings"
"time"
"craftdesign.group/quarry/internal/obs"
duckdb "github.com/marcboeker/go-duckdb/v2"
)
// Traffic answers the traffic page from the request log.
//
// The log holds facts and the page wants conclusions, so the first step
// is the one that turns one into the other: every distinct combination
// of the facts obs.Classify reads is classified once, in Go, into a
// table the rest of the queries join. There are a few thousand such
// combinations in a month, against hundreds of thousands of rows, and
// it keeps the rule in one language — the same function decides a row
// whichever window it is read in, so a better rule rereads the history.
func (s *Sink) Traffic(q obs.TrafficQuery) (*obs.Traffic, error) {
s.Flush()
ctx, cancel := context.WithTimeout(s.ctx, 20*time.Second)
defer cancel()
// One connection for the whole read: the classification table is a
// temporary one, and a temporary table is the connection's.
conn, err := s.db.Conn(ctx)
if err != nil {
return nil, err
}
defer conn.Close()
now := q.Now
if now.IsZero() {
now = time.Now()
}
rg := q.Range
if rg.Step <= 0 {
rg = obs.RangeNamed("")
}
n := int(rg.Span / rg.Step)
to := now.Truncate(rg.Step).Add(rg.Step)
from := to.Add(-time.Duration(n) * rg.Step)
prev := from.Add(-rg.Span)
t := &obs.Traffic{Range: rg, From: from, To: to}
if err := classifyInto(ctx, conn, prev, to); err != nil {
return nil, fmt.Errorf("obsduck: traffic: classify: %w", err)
}
where, fargs := filterSQL(q.Filters)
// base is the window's rows with their conclusions joined on. Every
// panel reads it, so every filter narrows every panel.
//
// It is built once per window into a temporary table, which every
// panel then reads: the page asks some eighteen questions of the
// window, and each one re-joined the raw log to the conclusions on
// five columns and re-scanned it.
base := func(name string, lo, hi time.Time) (string, []any, error) {
sel := `
SELECT hit.ts, coalesce(hit.door, '') AS door, coalesce(hit.url, '') AS url,
coalesce(hit.repo, '') AS repo, coalesce(hit.status, 0) AS status,
coalesce(hit.bytes, 0) AS bytes, coalesce(hit.elapsed_ms, 0) AS elapsed_ms,
coalesce(hit.stream, false) AS stream, coalesce(hit.country, '') AS country,
coalesce(hit.referrer, '') AS referrer, coalesce(hit.account, '') AS account,
coalesce(hit.verb, '') AS verb, coalesce(hit.asn, CAST(0 AS UINTEGER)) AS asn,
coalesce(hit.network, '') AS network, hit.visitor,
k.kind, k.cli, k.crawler, k.verified, k.impostor
FROM hit JOIN k ON coalesce(hit.ua, '') = k.ua
AND (coalesce(hit.account, '') <> '') = k.acct
AND coalesce(hit.door, '') = k.door
AND coalesce(hit.client, '') = k.client
AND coalesce(hit.asn, CAST(0 AS UINTEGER)) = k.asn
WHERE hit.ts >= ? AND hit.ts < ?` + where
if _, err := conn.ExecContext(ctx, `CREATE OR REPLACE TEMP TABLE `+name+` AS `+sel, append([]any{lo, hi}, fargs...)...); err != nil {
return "", nil, fmt.Errorf("obsduck: traffic: window: %w", err)
}
return `WITH h AS (SELECT * FROM ` + name + `) `, nil, nil
}
cur, args, err := base("hcur", from, to)
if err != nil {
return nil, err
}
if t.Totals, err = totals(ctx, conn, cur, args); err != nil {
return nil, fmt.Errorf("obsduck: traffic: totals: %w", err)
}
pb, pargs, err := base("hprev", prev, from)
if err != nil {
return nil, err
}
if t.Prev, err = totals(ctx, conn, pb, pargs); err != nil {
return nil, err
}
if t.Points, err = points(ctx, conn, cur, args, from, rg.Step, n); err != nil {
return nil, fmt.Errorf("obsduck: traffic: points: %w", err)
}
if t.PrevPoints, err = points(ctx, conn, pb, pargs, prev, rg.Step, n); err != nil {
return nil, err
}
counts := []struct {
dst *[]obs.Count
sql string
}{
{&t.Kinds, `SELECT kind, '', count(*), kind, false FROM h GROUP BY kind ORDER BY 3 DESC`},
// Impostors first: a crawler's name sent from the wrong network
// is the row on this card worth reading.
{&t.Crawlers, `SELECT crawler, mode(network), count(*), crawler, impostor FROM h
WHERE crawler <> '' GROUP BY crawler, impostor ORDER BY impostor DESC, 3 DESC LIMIT 12`},
{&t.Networks, `SELECT mode(network), 'AS' || asn::VARCHAR || CASE WHEN mode(country) <> '' THEN ' · ' || mode(country) ELSE '' END,
count(*), asn::VARCHAR, false FROM h WHERE asn <> 0 GROUP BY asn ORDER BY 3 DESC LIMIT 12`},
{&t.Countries, `SELECT country, '', count(*), country, false FROM h
WHERE country <> '' GROUP BY country ORDER BY 3 DESC LIMIT 250`},
{&t.Paths, `SELECT url, '', count(*), url, false FROM h
WHERE door IN ('web', 'api') AND NOT stream GROUP BY url ORDER BY 3 DESC LIMIT 40`},
{&t.Served, `SELECT ` + servedSQL + `, '', sum(bytes), '', false FROM h GROUP BY 1 HAVING sum(bytes) > 0 ORDER BY 3 DESC`},
{&t.Status, `SELECT status::VARCHAR, '', count(*), status::VARCHAR, false FROM h
WHERE status > 0 GROUP BY status ORDER BY 3 DESC LIMIT 8`},
{&t.Referrers, `SELECT referrer, '', count(*), referrer, false FROM h
WHERE referrer <> '' AND kind = 'person' GROUP BY referrer ORDER BY 3 DESC LIMIT 15`},
{&t.Refused, `SELECT door, CASE WHEN mode(repo) <> '' THEN 'mostly ' || mode(repo) ELSE '' END, count(*), door, false FROM h
WHERE status IN (429, 503) GROUP BY door ORDER BY 3 DESC`},
{&t.How, `SELECT CASE WHEN url LIKE '/r/%/archive/%' THEN 'archive' ELSE cli || ' ' || verb END,
'', count(*), '', false FROM h WHERE ` + downloadSQL + ` GROUP BY 1 ORDER BY 3 DESC LIMIT 10`},
}
for _, c := range counts {
if *c.dst, err = countRows(ctx, conn, cur+c.sql, args); err != nil {
return nil, fmt.Errorf("%.60s: %w", strings.Join(strings.Fields(c.sql), " "), err)
}
}
for i := range t.Status {
if code, err := strconv.Atoi(t.Status[i].Label); err == nil {
t.Status[i].Label = strconv.Itoa(code) + " " + http.StatusText(code)
}
}
for i := range t.Crawlers {
if t.Crawlers[i].Flag {
t.Crawlers[i].Label = "claims to be " + t.Crawlers[i].Label
t.Crawlers[i].Note = "from " + t.Crawlers[i].Note
}
}
if t.Slowest, err = latencies(ctx, conn, cur+`SELECT url, count(*),
CAST(quantile_disc(elapsed_ms, 0.5) AS BIGINT), CAST(quantile_disc(elapsed_ms, 0.95) AS BIGINT)
FROM h WHERE door IN ('web', 'api') AND NOT stream
GROUP BY url HAVING count(*) >= 3 ORDER BY 4 DESC LIMIT 40`, args); err != nil {
return nil, fmt.Errorf("slowest: %w", err)
}
if t.ByKind, err = latencies(ctx, conn, cur+`SELECT kind, count(*),
CAST(quantile_disc(elapsed_ms, 0.5) AS BIGINT), CAST(quantile_disc(elapsed_ms, 0.95) AS BIGINT)
FROM h WHERE NOT stream GROUP BY kind ORDER BY 1`, args); err != nil {
return nil, fmt.Errorf("by kind: %w", err)
}
if err := histogram(ctx, conn, cur, args, t); err != nil {
return nil, fmt.Errorf("obsduck: traffic: histogram: %w", err)
}
if err := conn.QueryRowContext(ctx, `SELECT coalesce(sum(zit_ms), 0), coalesce(sum(ops_ms), 0), coalesce(sum(elapsed_ms), 0)
FROM render WHERE ts >= ? AND ts < ?`, from, to).Scan(&t.StoreMs, &t.LogicMs, &t.RenderMs); err != nil {
return nil, fmt.Errorf("time split: %w", err)
}
// ops contains zit, and elapsed contains ops: split them into parts
// that add up, as the panel draws them.
t.RenderMs = max(0, t.RenderMs-t.LogicMs)
t.LogicMs = max(0, t.LogicMs-t.StoreMs)
if t.Downloads, err = downloads(ctx, conn, cur, args); err != nil {
return nil, fmt.Errorf("downloads: %w", err)
}
if t.Sources, err = sources(ctx, conn, cur, args); err != nil {
return nil, fmt.Errorf("sources: %w", err)
}
return t, nil
}
// downloadSQL is what counts as a download: a sync round that reads,
// and an archive, either one answered. A refused request is no download:
// probes of a name that is no repository (404s to /r/zqxwvut/archive/...)
// were listed as 13 archives of a repository that does not exist. An
// archive's 302 to the bucket is the download handed on, so it counts.
const downloadSQL = `(status < 400 AND ((door = 'sync' AND verb IN ('clone', 'fetch')) OR url LIKE '/r/%/archive/%'))`
// servedSQL names what a response was, for the data-served card.
const servedSQL = `CASE
WHEN door = 'sync' THEN 'sync and clone'
WHEN url LIKE '/r/%/archive/%' THEN 'archives'
WHEN url LIKE '/r/%/file/%' OR url LIKE '/r/%/doc/%' THEN 'files'
WHEN door = 'api' THEN 'API'
WHEN door = 'asset' THEN 'stylesheets, scripts and icons'
WHEN door = 'runner' THEN 'runner downloads'
ELSE 'pages' END`
// filters maps a filter chip to the column it narrows.
var filters = map[string]string{
"repo": "hit.repo", "kind": "k.kind", "network": "hit.asn",
"country": "hit.country", "path": "hit.url", "client": "k.cli",
"crawler": "k.crawler", "door": "hit.door",
}
func filterSQL(f map[string]string) (string, []any) {
var b strings.Builder
var args []any
for _, name := range obs.TrafficFilters {
v, ok := f[name]
if !ok || v == "" {
continue
}
if name == "network" {
// An AS number, compared as one: the column is unsigned and
// a string compared to it is no network at all.
n, err := strconv.ParseUint(v, 10, 32)
if err != nil {
n = 0
}
b.WriteString(" AND hit.asn = CAST(? AS UINTEGER)")
args = append(args, uint32(n))
continue
}
b.WriteString(" AND " + filters[name] + " = ?")
args = append(args, v)
}
return b.String(), args
}
// classifyInto fills the connection's classification table for every
// combination of facts the window holds.
func classifyInto(ctx context.Context, conn *sql.Conn, lo, hi time.Time) error {
for _, ddl := range []string{
`CREATE TEMP TABLE IF NOT EXISTS k (ua VARCHAR, acct BOOLEAN, door VARCHAR, client VARCHAR,
asn UINTEGER, kind VARCHAR, cli VARCHAR, crawler VARCHAR, verified BOOLEAN, impostor BOOLEAN)`,
`DELETE FROM k`,
} {
if _, err := conn.ExecContext(ctx, ddl); err != nil {
return fmt.Errorf("obsduck: traffic: %w", err)
}
}
rows, err := conn.QueryContext(ctx, `SELECT DISTINCT coalesce(ua, ''), coalesce(account, '') <> '',
coalesce(door, ''), coalesce(client, ''), coalesce(asn, CAST(0 AS UINTEGER))
FROM hit WHERE ts >= ? AND ts < ?`, lo, hi)
if err != nil {
return fmt.Errorf("obsduck: traffic: %w", err)
}
type fact struct {
ua, door, client string
acct bool
asn uint32
}
var facts []fact
for rows.Next() {
var f fact
if err := rows.Scan(&f.ua, &f.acct, &f.door, &f.client, &f.asn); err != nil {
rows.Close()
return err
}
facts = append(facts, f)
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
// Through the appender: rows go to DuckDB in bulk with no SQL to bind.
// One INSERT per fact cost 17 of the 7-day page's 20 seconds, and 500
// to a statement no better, since go-duckdb looks each of a
// statement's parameters up by name across cgo (ParameterName,
// ParamType): binding, not inserting, was the cost.
return conn.Raw(func(dc any) error {
a, err := duckdb.NewAppender(dc.(driver.Conn), "temp", "main", "k")
if err != nil {
return fmt.Errorf("obsduck: traffic: appender: %w", err)
}
for _, f := range facts {
acct := ""
if f.acct {
acct = "yes"
}
r := obs.Classify(obs.Facts{UA: f.ua, Account: acct, Door: f.door, Client: f.client, ASN: f.asn})
if err := a.AppendRow(f.ua, f.acct, f.door, f.client, f.asn,
r.Kind, r.Client, r.Crawler, r.Verified, r.Impostor); err != nil {
a.Close()
return fmt.Errorf("obsduck: traffic: append: %w", err)
}
}
return a.Close()
})
}
func totals(ctx context.Context, conn *sql.Conn, base string, args []any) (obs.TrafficTotals, error) {
var t obs.TrafficTotals
var auto int64
err := conn.QueryRowContext(ctx, base+`SELECT count(*), count(DISTINCT visitor), coalesce(sum(bytes), 0),
coalesce(sum(CASE WHEN NOT stream THEN elapsed_ms END), 0),
count(*) FILTER (WHERE status >= 500),
count(*) FILTER (WHERE kind IN ('crawler', 'automated'))
FROM h`, args...).Scan(&t.Requests, &t.Visitors, &t.Bytes, &t.ServerMs, &t.Errors, &auto)
if t.Requests > 0 {
t.Automated = float64(auto) / float64(t.Requests)
}
return t, err
}
// points is the chart: n points of step from lo, every one present
// whether or not anything happened in it, so a quiet hour reads as a
// zero rather than as a gap the line jumps over.
func points(ctx context.Context, conn *sql.Conn, base string, args []any, lo time.Time, step time.Duration, n int) ([]obs.TrafficPoint, error) {
out := make([]obs.TrafficPoint, n)
for i := range out {
out[i].At = lo.Add(time.Duration(i) * step)
}
idx := `CAST(floor((epoch(ts) - ?) / ?) AS BIGINT)`
a := append([]any{}, args...)
a = append(a, float64(lo.Unix()), step.Seconds())
rows, err := conn.QueryContext(ctx, base+`SELECT `+idx+` AS i, kind, count(*) FROM h GROUP BY i, kind`, a...)
if err != nil {
return nil, err
}
for rows.Next() {
var i, c int64
var kind string
if err := rows.Scan(&i, &kind, &c); err != nil {
rows.Close()
return nil, err
}
if i < 0 || i >= int64(n) {
continue
}
for j, k := range obs.TrafficKinds {
if k == kind {
out[i].ByKind[j] += c
}
}
}
rows.Close()
rows, err = conn.QueryContext(ctx, base+`SELECT `+idx+` AS i, count(DISTINCT visitor), coalesce(sum(bytes), 0),
coalesce(sum(CASE WHEN NOT stream THEN elapsed_ms END), 0), count(*) FILTER (WHERE status >= 500)
FROM h GROUP BY i`, a...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var i int64
var p obs.TrafficPoint
if err := rows.Scan(&i, &p.Visitors, &p.Bytes, &p.ServerMs, &p.Errors); err != nil {
return nil, err
}
if i < 0 || i >= int64(n) {
continue
}
out[i].Visitors, out[i].Bytes, out[i].ServerMs, out[i].Errors = p.Visitors, p.Bytes, p.ServerMs, p.Errors
}
return out, rows.Err()
}
func countRows(ctx context.Context, conn *sql.Conn, q string, args []any) ([]obs.Count, error) {
rows, err := conn.QueryContext(ctx, q, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: traffic: %w", err)
}
defer rows.Close()
var out []obs.Count
for rows.Next() {
var c obs.Count
var label, note, key sql.NullString
if err := rows.Scan(&label, ¬e, &c.N, &key, &c.Flag); err != nil {
return nil, err
}
c.Label, c.Note, c.Key = label.String, note.String, key.String
out = append(out, c)
}
return out, rows.Err()
}
func latencies(ctx context.Context, conn *sql.Conn, q string, args []any) ([]obs.Latency, error) {
rows, err := conn.QueryContext(ctx, q, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: traffic: %w", err)
}
defer rows.Close()
var out []obs.Latency
for rows.Next() {
var l obs.Latency
if err := rows.Scan(&l.Label, &l.N, &l.P50Ms, &l.P95Ms); err != nil {
return nil, err
}
out = append(out, l)
}
return out, rows.Err()
}
func histogram(ctx context.Context, conn *sql.Conn, base string, args []any, t *obs.Traffic) error {
var b strings.Builder
b.WriteString("CASE")
for i, e := range obs.HistEdgesMs {
fmt.Fprintf(&b, " WHEN elapsed_ms < %d THEN %d", e, i)
}
fmt.Fprintf(&b, " ELSE %d END", len(obs.HistEdgesMs))
t.Histogram = make([]int64, len(obs.HistEdgesMs)+1)
rows, err := conn.QueryContext(ctx, base+`SELECT `+b.String()+`, count(*) FROM h WHERE NOT stream GROUP BY 1`, args...)
if err != nil {
return err
}
for rows.Next() {
var i, c int64
if err := rows.Scan(&i, &c); err != nil {
rows.Close()
return err
}
t.Histogram[i] = c
}
rows.Close()
var p50, p95, p99 sql.NullInt64
err = conn.QueryRowContext(ctx, base+`SELECT CAST(quantile_disc(elapsed_ms, 0.5) AS BIGINT),
CAST(quantile_disc(elapsed_ms, 0.95) AS BIGINT), CAST(quantile_disc(elapsed_ms, 0.99) AS BIGINT)
FROM h WHERE NOT stream`, args...).Scan(&p50, &p95, &p99)
t.P50, t.P95, t.P99 = p50.Int64, p95.Int64, p99.Int64
return err
}
func downloads(ctx context.Context, conn *sql.Conn, base string, args []any) ([]obs.Download, error) {
rows, err := conn.QueryContext(ctx, base+`SELECT repo,
count(*) FILTER (WHERE verb = 'clone'), count(*) FILTER (WHERE verb = 'fetch'),
count(*) FILTER (WHERE url LIKE '/r/%/archive/%'), sum(bytes),
count(*) FILTER (WHERE kind = 'person'), count(*) FILTER (WHERE kind = 'crawler'),
count(*) FILTER (WHERE kind = 'automated'), count(*) FILTER (WHERE kind = 'unsure'),
count(*) FILTER (WHERE cli = 'zit'), count(*) FILTER (WHERE cli = 'fossil'),
count(*) FILTER (WHERE cli NOT IN ('zit', 'fossil')),
count(DISTINCT account) FILTER (WHERE account <> '')
FROM h WHERE repo <> '' AND `+downloadSQL+`
GROUP BY repo ORDER BY count(*) DESC LIMIT 30`, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: traffic: %w", err)
}
defer rows.Close()
var out []obs.Download
for rows.Next() {
var d obs.Download
if err := rows.Scan(&d.Repo, &d.Clones, &d.Fetches, &d.Archive, &d.Bytes,
&d.Mix[0], &d.Mix[1], &d.Mix[2], &d.Mix[3], &d.Zit, &d.Fossil, &d.Other, &d.SignedIn); err != nil {
return nil, err
}
out = append(out, d)
}
return out, rows.Err()
}
// sources groups the downloads by who they appear to come from: every
// signed-in account together, and everyone else by network and client.
// The evidence for each conclusion is put into words, because a reader
// deciding whether to block a network wants to know why it reads as a
// mirror, not only that it does.
func sources(ctx context.Context, conn *sql.Conn, base string, args []any) ([]obs.Source, error) {
rows, err := conn.QueryContext(ctx, base+`, d0 AS (
SELECT *, CASE WHEN account <> '' THEN '' ELSE network || ' ' || cli END AS src
FROM h WHERE `+downloadSQL+`),
d AS (SELECT *, epoch(ts) - lag(epoch(ts)) OVER (PARTITION BY src ORDER BY ts) AS gap FROM d0)
SELECT d.src, count(*), mode(kind), mode(cli), mode(network),
string_agg(DISTINCT repo, ', ' ORDER BY repo),
string_agg(DISTINCT CASE WHEN url LIKE '/r/%/archive/%' THEN 'archive' ELSE verb END, ' and '),
coalesce(string_agg(DISTINCT nullif(account, ''), ', '), ''),
count(DISTINCT account) FILTER (WHERE account <> ''),
count(DISTINCT visitor), bool_or(impostor), bool_or(verified), mode(crawler),
coalesce(quantile_cont(gap, 0.5), 0),
CAST(epoch(max(ts)) - epoch(min(ts)) AS BIGINT)
FROM d GROUP BY d.src ORDER BY 2 DESC LIMIT 12`, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: traffic: %w", err)
}
defer rows.Close()
var out []obs.Source
for rows.Next() {
var (
src, kind, cli, network, repos, verbs, accounts, crawler string
n, nAccounts, visitors, spanS int64
impostor, verified bool
gap float64
)
if err := rows.Scan(&src, &n, &kind, &cli, &network, &repos, &verbs, &accounts, &nAccounts,
&visitors, &impostor, &verified, &crawler, &gap, &spanS); err != nil {
return nil, err
}
s := obs.Source{Kind: kind, N: n, Took: firstOf(repos, 3) + " · " + verbs}
switch {
case src == "":
s.Name = firstOf(accounts, 2)
s.Why = "signed in"
s.Kind = obs.KindPerson
default:
s.Name = sourceName(network, cli, visitors)
s.Why = sourceWhy(kind, network, crawler, impostor, verified, n, visitors, gap, spanS)
}
out = append(out, s)
}
return out, rows.Err()
}
// firstOf is a comma list cut to its first n names and a count of the
// rest.
func firstOf(list string, n int) string {
names := strings.Split(list, ", ")
if len(names) <= n {
return list
}
if rest := len(names) - n; rest == 1 {
return strings.Join(names[:n], ", ") + " and 1 other"
}
return strings.Join(names[:n], ", ") + fmt.Sprintf(" and %d others", len(names)-n)
}
func sourceName(network, cli string, visitors int64) string {
where := network
if where == "" {
where = "an unknown network"
}
if visitors == 1 {
return "one " + where + " address, " + cli
}
return fmt.Sprintf("%d readers on %s, %s", visitors, where, cli)
}
// sourceWhy puts the evidence for a source's kind into a sentence.
func sourceWhy(kind, network, crawler string, impostor, verified bool, n, visitors int64, gap float64, spanS int64) string {
switch {
case impostor:
return "claims to be " + crawler + " from a network that is not " + crawler + "'s"
case verified:
return crawler + ", verified by its network"
case n >= 20 && gap > 0 && visitors <= 2 && float64(spanS) > 6*3600:
return "comes back every " + every(gap) + " and never signs in: a mirror or a scheduled job"
case kind == obs.KindAutomated:
return "a program's user agent, not a browser's or a sync client's"
case visitors > 1:
return fmt.Sprintf("%d different readers, irregular gaps", visitors)
}
return "one reader on " + network + "; nothing yet says person or program"
}
func every(s float64) string {
switch {
case s < 90:
return fmt.Sprintf("%.0fs", math.Round(s))
case s < 5400:
return fmt.Sprintf("%.0f minutes", math.Round(s/60))
}
return fmt.Sprintf("%.0f hours", math.Round(s/3600))
}