Merge branch 'feature/traffic-analytics'

db.go

@@ -44,6 +44,16 @@ CREATE TABLE IF NOT EXISTS traffic_hourly (
 	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
 	PRIMARY KEY (mapping, hour)
 );
+CREATE TABLE IF NOT EXISTS traffic_minutely (
+	mapping TEXT NOT NULL, minute INTEGER NOT NULL, conns INTEGER NOT NULL DEFAULT 0,
+	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
+	PRIMARY KEY (mapping, minute)
+);
+CREATE TABLE IF NOT EXISTS traffic_sources (
+	mapping TEXT NOT NULL, hour INTEGER NOT NULL, ip TEXT NOT NULL, conns INTEGER NOT NULL DEFAULT 0,
+	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
+	PRIMARY KEY (mapping, hour, ip)
+);
 CREATE TABLE IF NOT EXISTS alerts (
 	key TEXT PRIMARY KEY, org_id TEXT NOT NULL DEFAULT '', kind TEXT, name TEXT, message TEXT,
 	since INTEGER NOT NULL, fired INTEGER NOT NULL DEFAULT 0, notified INTEGER NOT NULL DEFAULT 0
@@ -555,16 +565,160 @@ func (d *DB) purge(accessDays, auditDays int) {
 	if auditDays > 0 {
 		d.sql.Exec(`DELETE FROM audit_log WHERE ts < ?`, ms(time.Now().AddDate(0, 0, -auditDays)))
 	}
-	d.sql.Exec(`DELETE FROM traffic_hourly WHERE hour < ?`, time.Now().Add(-31*24*time.Hour).Unix()/3600)
+	d.sql.Exec(`DELETE FROM traffic_hourly WHERE hour < ?`, time.Now().Add(-trafficHourlyKeep).Unix()/3600)
+	d.sql.Exec(`DELETE FROM traffic_minutely WHERE minute < ?`, time.Now().Add(-trafficMinutelyKeep).Unix()/60)
+	d.sql.Exec(`DELETE FROM traffic_sources WHERE hour < ?`, time.Now().Add(-trafficSourcesKeep).Unix()/3600)
 	d.sql.Exec(`DELETE FROM alert_events WHERE ts < ?`, ms(time.Now().AddDate(0, 0, -90)))
 }
 
 // ---------- traffic history ----------
 
-func (d *DB) addTraffic(mapping string, st model.Stat) {
+// How long each resolution of traffic history is kept. Minutes are for the last hours, hours for the
+// months, and the busiest client addresses sit in between.
+const (
+	trafficMinutelyKeep = 48 * time.Hour
+	trafficSourcesKeep  = 7 * 24 * time.Hour
+	trafficHourlyKeep   = 31 * 24 * time.Hour
+)
+
+func (d *DB) addTraffic(mapping string, st model.Stat) { d.addTrafficAt(mapping, st, time.Now()) }
+
+func (d *DB) addTrafficAt(mapping string, st model.Stat, at time.Time) {
 	d.sql.Exec(`INSERT INTO traffic_hourly (mapping, hour, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?)
 		ON CONFLICT (mapping, hour) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
-		mapping, time.Now().Unix()/3600, st.Conns, st.BytesIn, st.BytesOut)
+		mapping, at.Unix()/3600, st.Conns, st.BytesIn, st.BytesOut)
+	d.sql.Exec(`INSERT INTO traffic_minutely (mapping, minute, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?)
+		ON CONFLICT (mapping, minute) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
+		mapping, at.Unix()/60, st.Conns, st.BytesIn, st.BytesOut)
+	for _, src := range st.Sources {
+		d.sql.Exec(`INSERT INTO traffic_sources (mapping, hour, ip, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?, ?)
+			ON CONFLICT (mapping, hour, ip) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
+			mapping, at.Unix()/3600, src.IP, src.Conns, src.BytesIn, src.BytesOut)
+	}
+}
+
+// TrafficPoint is the traffic of one time bucket.
+type TrafficPoint struct {
+	Time     int64 `json:"t"` // unix seconds at the start of the bucket
+	Conns    int64 `json:"conns"`
+	BytesIn  int64 `json:"bytesIn"`
+	BytesOut int64 `json:"bytesOut"`
+}
+
+// trafficSeries sums the given ports' traffic into count buckets of the given size ("minute" or
+// "hour") ending with the current one, oldest first. Empty buckets are included.
+func (d *DB) trafficSeries(ids []string, unit string, count int, now time.Time) []TrafficPoint {
+	table, col, secs := "traffic_hourly", "hour", int64(3600)
+	if unit == "minute" {
+		table, col, secs = "traffic_minutely", "minute", 60
+	}
+	cur := now.Unix() / secs
+	first := cur - int64(count) + 1
+	out := make([]TrafficPoint, count)
+	for i := range out {
+		out[i].Time = (first + int64(i)) * secs
+	}
+	if len(ids) == 0 {
+		return out
+	}
+	q := `SELECT ` + col + `, SUM(conns), SUM(bytes_in), SUM(bytes_out) FROM ` + table +
+		` WHERE ` + col + ` >= ? AND ` + col + ` <= ? AND mapping IN (` + placeholders(len(ids)) + `) GROUP BY ` + col
+	args := []any{first, cur}
+	for _, id := range ids {
+		args = append(args, id)
+	}
+	rows, err := d.sql.Query(q, args...)
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var at int64
+		var p TrafficPoint
+		if rows.Scan(&at, &p.Conns, &p.BytesIn, &p.BytesOut) != nil {
+			continue
+		}
+		if i := int(at - first); i >= 0 && i < count {
+			out[i].Conns, out[i].BytesIn, out[i].BytesOut = p.Conns, p.BytesIn, p.BytesOut
+		}
+	}
+	return out
+}
+
+func placeholders(n int) string {
+	if n <= 0 {
+		return ""
+	}
+	return strings.TrimSuffix(strings.Repeat("?,", n), ",")
+}
+
+// TrafficSource is one client address's traffic over a period.
+type TrafficSource struct {
+	IP       string `json:"ip"`
+	Conns    int64  `json:"conns"`
+	BytesIn  int64  `json:"bytesIn"`
+	BytesOut int64  `json:"bytesOut"`
+}
+
+// topSources lists the busiest client addresses of the given ports over the last hours hours.
+func (d *DB) topSources(ids []string, hours, limit int, now time.Time) []TrafficSource {
+	out := []TrafficSource{}
+	if len(ids) == 0 {
+		return out
+	}
+	args := []any{now.Unix()/3600 - int64(hours) + 1}
+	for _, id := range ids {
+		args = append(args, id)
+	}
+	args = append(args, limit)
+	rows, err := d.sql.Query(`SELECT ip, SUM(conns), SUM(bytes_in), SUM(bytes_out) FROM traffic_sources WHERE hour >= ? AND mapping IN (`+placeholders(len(ids))+`)
+		GROUP BY ip ORDER BY SUM(bytes_in) + SUM(bytes_out) DESC, SUM(conns) DESC, ip LIMIT ?`, args...)
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var s TrafficSource
+		if rows.Scan(&s.IP, &s.Conns, &s.BytesIn, &s.BytesOut) == nil {
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+// bytesSince is the total bytes (in + out) a port has carried from the start of the given hour (unix hours) until now.
+func (d *DB) bytesSince(mapping string, fromHour int64) int64 {
+	var n sql.NullInt64
+	d.sql.QueryRow(`SELECT SUM(bytes_in + bytes_out) FROM traffic_hourly WHERE mapping = ? AND hour >= ?`, mapping, fromHour).Scan(&n)
+	return n.Int64
+}
+
+// spikeWindow and spikeBaseline are the minutes a spike is judged over, and the minutes before it
+// that say what normal looks like.
+const (
+	spikeWindow   = 5
+	spikeBaseline = 6 * 60
+)
+
+// recentTraffic returns the bytes a port carried in the last spikeWindow minutes, the average for
+// the same span over the baseline before it, and how many minutes that baseline covers (up to spikeBaseline;
+// fewer for a port with a shorter history, which is averaged over what it has).
+func (d *DB) recentTraffic(mapping string, now time.Time) (recent int64, baselinePerWindow float64, historyMinutes int64) {
+	cur := now.Unix() / 60
+	var r, b, first sql.NullInt64
+	d.sql.QueryRow(`SELECT SUM(CASE WHEN minute > ? THEN bytes_in + bytes_out END), SUM(CASE WHEN minute <= ? THEN bytes_in + bytes_out END), MIN(minute)
+		FROM traffic_minutely WHERE mapping = ? AND minute > ?`, cur-spikeWindow, cur-spikeWindow, mapping, cur-spikeWindow-spikeBaseline).Scan(&r, &b, &first)
+	if !first.Valid {
+		return 0, 0, 0
+	}
+	span := cur - spikeWindow - first.Int64 + 1
+	if span > spikeBaseline {
+		span = spikeBaseline
+	}
+	if span <= 0 {
+		return r.Int64, 0, 0
+	}
+	return r.Int64, float64(b.Int64) / float64(span) * spikeWindow, span
 }
 
 // trafficHistory returns bytes (in + out) per hour for each mapping, oldest first.