diff --git a/README.md b/README.md index ac963ba..cafdc7c 100644 --- a/README.md +++ b/README.md @@ -171,7 +171,9 @@ for each provider (e.g. `CF_DNS_API_TOKEN` for `cloudflare`); `.env` is the plac - Several targets per port: round-robin load balancing with automatic failover. - Health checks (TCP connect or an HTTP path); unhealthy targets are skipped and shown in the UI. - Conflict detection (same ip/port/protocol, overlapping ranges, or `0.0.0.0` overlapping a specific IP). -- Traffic totals and a 24-hour sparkline per port. +- Traffic totals and a 24-hour sparkline per port, plus a **Traffic** chart per resource (and a + **Traffic overview** for the whole organization) with bytes in and out, connections and the + busiest visitor addresses. See *Traffic analytics* below. **HTTPS**: automatic Let's Encrypt certificates, optional port-80 redirect, `X-Forwarded-*` headers, Host rewrite, HTTPS to the backend, certificate status per port. @@ -212,6 +214,24 @@ headers, Host rewrite, HTTPS to the backend, certificate status per port. - **`/metrics`** (Prometheus) when `HAMMERHEAD_METRICS_TOKEN` is set: edge/site/port/target/mail-check gauges, traffic counters, alerts firing. Off (404) otherwise. +**Traffic analytics** +- Every edge counts connections and bytes in and out per port, plus the ten busiest client addresses + per port, and sends them with its regular report (every 20 seconds). Hammerhead adds them up at + three resolutions: per minute (kept 48 hours), per hour (31 days) and, for visitor addresses, per + hour (7 days). Older rows are deleted by the same hourly clean-up as the logs. +- The charts are on the Resources page: **Traffic** on a row, or **Traffic overview** for everything. + Pick an hour or 6 hours (per minute), a day, a week or a month (per hour). Visitor addresses cover + at most the last 7 days, and very small visitors may be missing because each report names ten. +- `GET /api/orgs/{org}/traffic/{resource id or "all"}?range=1h|6h|24h|7d|30d` returns the series, + totals and busiest addresses as JSON, for scripts or your own dashboards. +- **Traffic alerts** (Alerts → Traffic, per organization, off by default), delivered like every other + alert, with the same debounce: + - *Spike*: a port carried at least N times its usual traffic in the last 5 minutes, compared with + the average over the previous 6 hours. A port needs an hour of history, and the burst must be over + 10 MB, so quiet ports don't page you. + - *Daily amount*: a port carried more than the set number of MB (in plus out) since midnight UTC. + Neither blocks traffic; they only tell you. + **Mail egress (send your mail server's outgoing mail from the edge's IP)** - Receiving servers judge mail by the IP it comes from: its reverse DNS, your HELO name and your SPF record all have to agree. A mail server at home usually can't satisfy that (residential IP, port 25 diff --git a/alerts.go b/alerts.go index 04cfe65..6f56eef 100644 --- a/alerts.go +++ b/alerts.go @@ -127,6 +127,7 @@ func (s *Server) problemsLocked(now time.Time) map[string]problem { } } } + s.trafficProblemsLocked(now, edgeDown, add) for _, x := range s.st.MailEgress { if !x.Enabled || edgeDown[x.EdgeID] || siteDown[x.SiteID] { continue diff --git a/alerts_api.go b/alerts_api.go index 656e319..f3ddb02 100644 --- a/alerts_api.go +++ b/alerts_api.go @@ -61,12 +61,13 @@ func (s *Server) alertOverview(w http.ResponseWriter, sc alertScope) { } } sort.Slice(active, func(i, j int) bool { return active[i].Since.Before(active[j].Since) }) - settings := map[string]int{"debounceMin": 2, "repeatHours": 0} + settings := map[string]int{"debounceMin": 2, "repeatHours": 0, "spikeFactor": 0, "dailyMb": 0} if o := s.findOrg(sc.org); o != nil { if o.AlertDebounceMin > 0 { settings["debounceMin"] = o.AlertDebounceMin } settings["repeatHours"] = o.AlertRepeatHours + settings["spikeFactor"], settings["dailyMb"] = o.TrafficSpikeFactor, o.TrafficDailyMB } mailOK := s.st.Settings.SMTP.Host != "" s.mu.Unlock() @@ -198,6 +199,7 @@ func (s *Server) registerAlertRoutes(mux *http.ServeMux) { mux.HandleFunc("DELETE /api/orgs/{org}/alerts/channels/{id}", s.withOrgAlerts(s.deleteChannel)) mux.HandleFunc("POST /api/orgs/{org}/alerts/channels/{id}/test", s.withOrgAlerts(s.testChannel)) mux.HandleFunc("PUT /api/orgs/{org}/alerts/settings", s.withOrgAlerts(s.alertSettings)) + mux.HandleFunc("PUT /api/orgs/{org}/alerts/traffic", s.withOrgAlerts(s.trafficAlertSettings)) mux.HandleFunc("GET /api/admin/alerts", s.withServerAlerts(overview)) mux.HandleFunc("POST /api/admin/alerts/channels", s.withServerAlerts(s.saveChannel)) mux.HandleFunc("PUT /api/admin/alerts/channels/{id}", s.withServerAlerts(s.saveChannel)) diff --git a/db.go b/db.go index ea65bfa..c4de31c 100644 --- a/db.go +++ b/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 @@ -553,16 +563,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. diff --git a/e2e/traffic_test.go b/e2e/traffic_test.go new file mode 100644 index 0000000..0b5a2a2 --- /dev/null +++ b/e2e/traffic_test.go @@ -0,0 +1,80 @@ +package e2e + +import ( + "strconv" + "testing" + "time" + + "git.veillax.com/Veillax/blacktip" + "git.veillax.com/Veillax/reef/model" +) + +// Traffic through an edge shows up in Hammerhead's charts, with the client address named. +func TestTrafficAnalyticsFlow(t *testing.T) { + w := setup(t, blacktip.TraefikConfig{}) + target := freeRun(t, 1) + pub := freeRun(t, 1) + if pub == target { + pub = freeRun(t, 2) + 1 + } + tcpOn(t, target) + var m model.Mapping + if code, raw := w.admin.do("POST", "/api/orgs/"+w.org+"/mappings", map[string]any{"proto": "tcp", "edgeId": w.edge, "bindIp": "127.0.0.1", "port": pub, + "targets": tgt(w.local, target), "enabled": true}, &m); code != 200 { + t.Fatalf("open: %d %s", code, raw) + } + for i := 0; i < 3; i++ { + if got := tcpAsk(t, "127.0.0.1:"+strconv.Itoa(pub), "hello"); got == "" { + t.Fatal("no answer through the edge") + } + } + + type detail struct { + Total struct{ Conns, BytesIn, BytesOut int64 } `json:"total"` + Points []struct{ Conns int64 } `json:"points"` + Sources []struct { + IP string + Conns int64 + } `json:"sources"` + } + var d detail + for i := 0; ; i++ { // the edge reports every 20 seconds + d = detail{} + if code, _ := w.admin.do("GET", "/api/orgs/"+w.org+"/traffic/"+m.ID+"?range=1h", nil, &d); code == 200 && d.Total.Conns >= 3 { + break + } + if i >= 450 { + t.Fatal("timed out waiting for the edge to report the traffic") + } + time.Sleep(100 * time.Millisecond) + } + if d.Total.BytesIn < 15 || d.Total.BytesOut == 0 || len(d.Points) != 60 { + t.Errorf("totals: %+v with %d points", d.Total, len(d.Points)) + } + if len(d.Sources) != 1 || d.Sources[0].IP != "127.0.0.1" || d.Sources[0].Conns < 3 { + t.Errorf("sources: %+v", d.Sources) + } + + var all detail + if code, raw := w.admin.do("GET", "/api/orgs/"+w.org+"/traffic/all?range=24h", nil, &all); code != 200 || all.Total.Conns < 3 || len(all.Points) != 24 { + t.Errorf("organization total: %d %s", code, raw) + } + if code, _ := w.admin.do("GET", "/api/orgs/"+w.org+"/traffic/"+m.ID+"?range=5y", nil, nil); code != 400 { + t.Errorf("bad range: %d", code) + } + if code, _ := w.admin.do("GET", "/api/orgs/"+w.org+"/traffic/nope", nil, nil); code != 404 { + t.Errorf("unknown port: %d", code) + } + + alerts := "/api/orgs/" + w.org + "/alerts" + if code, _ := w.admin.do("PUT", alerts+"/traffic", map[string]any{"spikeFactor": 1}, nil); code != 400 { + t.Errorf("a spike of 1x was accepted: %d", code) + } + if code, raw := w.admin.do("PUT", alerts+"/traffic", map[string]any{"spikeFactor": 5, "dailyMb": 2048}, nil); code != 200 { + t.Fatalf("save traffic alerts: %d %s", code, raw) + } + var ov struct{ Settings map[string]int } + if code, _ := w.admin.do("GET", alerts, nil, &ov); code != 200 || ov.Settings["spikeFactor"] != 5 || ov.Settings["dailyMb"] != 2048 || ov.Settings["debounceMin"] != 2 { + t.Errorf("settings after save: %d %+v", code, ov.Settings) + } +} diff --git a/server.go b/server.go index 2b60aae..f57d304 100644 --- a/server.go +++ b/server.go @@ -149,6 +149,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/orgs/{org}/logs", s.withOrg(true, s.orgLogs)) mux.HandleFunc("GET /api/orgs/{org}/stats", s.withOrg(false, s.orgStats)) mux.HandleFunc("GET /api/orgs/{org}/traffic", s.withOrg(false, s.orgTraffic)) + mux.HandleFunc("GET /api/orgs/{org}/traffic/{id}", s.withOrg(false, s.orgTrafficDetail)) // agents mux.HandleFunc("GET /edge/config", s.edgeConfig) diff --git a/store.go b/store.go index 6a99b1c..138fc1b 100644 --- a/store.go +++ b/store.go @@ -29,6 +29,10 @@ type Org struct { AlertDebounceMin int `json:"alertDebounceMin,omitempty"` // minutes a problem must last before alerting; 0 = 2 AlertRepeatHours int `json:"alertRepeatHours,omitempty"` // remind about an unresolved problem this often; 0 = never + + // Traffic alerts, applied to every port of the organization. 0 = off. + TrafficSpikeFactor int `json:"trafficSpikeFactor,omitempty"` // alert when a port carries this many times its usual traffic + TrafficDailyMB int `json:"trafficDailyMb,omitempty"` // alert when a port carries more than this many MB in one UTC day } type Role struct { diff --git a/traffic.go b/traffic.go new file mode 100644 index 0000000..91dce15 --- /dev/null +++ b/traffic.go @@ -0,0 +1,161 @@ +package hammerhead + +import ( + "errors" + "fmt" + "net/http" + "time" +) + +// Traffic analytics: edges report per-port bytes, connections and their busiest client addresses with +// every report. Hammerhead keeps them at three resolutions (see the *Keep constants in db.go) and draws +// charts from them. Organizations can also be alerted when a port suddenly carries far more traffic +// than usual, or more than a daily amount. + +// Spikes are judged on the last few minutes against the hours before them. A port has to have an +// hour of history before it can spike, and the burst has to be worth noticing in absolute terms, so +// a quiet port going from 1 KB to 50 KB doesn't page anyone. +const ( + spikeMinHistory = 60 // minutes of baseline needed before a spike can be called + spikeMinBytes = 10 * 1024 * 1024 // bytes in the window below which nothing counts as a spike +) + +// spiking says whether recent traffic is a spike over the baseline for a port. +func spiking(recent int64, baseline float64, history int64, factor int) bool { + if factor <= 0 || history < spikeMinHistory || recent < spikeMinBytes { + return false + } + return float64(recent) > baseline*float64(factor) +} + +// trafficProblemsLocked adds an alert problem for each port of an organization that is spiking or has +// gone over its daily amount. Caller holds s.mu. +func (s *Server) trafficProblemsLocked(now time.Time, edgeDown map[string]bool, add func(key, org, kind, name, msg string)) { + dayStart := now.UTC().Truncate(24*time.Hour).Unix() / 3600 + for _, m := range s.st.Mappings { + if !m.Enabled || edgeDown[m.EdgeID] { + continue + } + o := s.findOrg(m.OrgID) + if o == nil || (o.TrafficSpikeFactor <= 0 && o.TrafficDailyMB <= 0) { + continue + } + if o.TrafficSpikeFactor > 0 { + recent, base, history := s.db.recentTraffic(m.ID, now) + if spiking(recent, base, history, o.TrafficSpikeFactor) { + add("spike:"+m.ID, m.OrgID, "traffic", m.Name, fmt.Sprintf("%s: %s in the last %d minutes, about %.0f times its usual %s", + m.Name, byteText(recent), spikeWindow, float64(recent)/maxFloat(base, 1), byteText(int64(base)))) + } + } + if limit := int64(o.TrafficDailyMB) << 20; limit > 0 { + if used := s.db.bytesSince(m.ID, dayStart); used > limit { + add("quota:"+m.ID, m.OrgID, "traffic", m.Name, fmt.Sprintf("%s has carried %s today, over the %s a day you set", m.Name, byteText(used), byteText(limit))) + } + } + } +} + +func maxFloat(a, b float64) float64 { + if a > b { + return a + } + return b +} + +func byteText(n int64) string { + f, units := float64(n), []string{"B", "KB", "MB", "GB", "TB"} + i := 0 + for f >= 1024 && i < len(units)-1 { + f /= 1024 + i++ + } + if i == 0 { + return fmt.Sprintf("%d B", n) + } + return fmt.Sprintf("%.1f %s", f, units[i]) +} + +// trafficRanges maps the ranges the dashboard offers to a bucket size, a bucket count and how many hours of top sources to include. +var trafficRanges = map[string]struct { + unit string + count int + hours int + bucket time.Duration +}{ + "1h": {"minute", 60, 1, time.Minute}, + "6h": {"minute", 360, 6, time.Minute}, + "24h": {"hour", 24, 24, time.Hour}, + "7d": {"hour", 7 * 24, 7 * 24, time.Hour}, + "30d": {"hour", 30 * 24, 7 * 24, time.Hour}, // client addresses are only kept for a week +} + +// orgTrafficDetail serves one port's traffic chart and busiest clients, or the organization's total +// for id "all". ?range= is one of 1h, 6h, 24h (default), 7d or 30d. +func (s *Server) orgTrafficDetail(w http.ResponseWriter, r *http.Request, c *orgCtx) { + rng := r.URL.Query().Get("range") + if rng == "" { + rng = "24h" + } + spec, found := trafficRanges[rng] + if !found { + fail(w, 400, errors.New("range must be 1h, 6h, 24h, 7d or 30d")) + return + } + id := r.PathValue("id") + s.mu.Lock() + var ids []string + name := "All resources" + for _, m := range s.st.Mappings { + if m.OrgID != c.Org.ID { + continue + } + if id == "all" { + ids = append(ids, m.ID) + } else if m.ID == id { + ids, name = []string{m.ID}, m.Name + } + } + s.mu.Unlock() + if id != "all" && len(ids) == 0 { + fail(w, 404, errors.New("no such resource")) + return + } + now := time.Now() + points := s.db.trafficSeries(ids, spec.unit, spec.count, now) + var tot TrafficPoint + var peak int64 + for _, p := range points { + tot.Conns += p.Conns + tot.BytesIn += p.BytesIn + tot.BytesOut += p.BytesOut + if b := p.BytesIn + p.BytesOut; b > peak { + peak = b + } + } + writeJSON(w, 200, map[string]any{ + "name": name, "range": rng, "bucketSeconds": int(spec.bucket / time.Second), "points": points, + "total": tot, "peakBucket": peak, "sources": s.db.topSources(ids, spec.hours, 10, now), "sourcesHours": spec.hours, + }) +} + +// trafficAlertSettings sets when an organization is told about unusual traffic. +func (s *Server) trafficAlertSettings(w http.ResponseWriter, r *http.Request, sc alertScope) { + var in struct{ SpikeFactor, DailyMB int } + if !decode(w, r, &in) { + return + } + if (in.SpikeFactor != 0 && (in.SpikeFactor < 2 || in.SpikeFactor > 1000)) || in.DailyMB < 0 || in.DailyMB > 100_000_000 { + fail(w, 400, errors.New("A spike is 2 to 1000 times the usual traffic (0 turns it off), and the daily amount can't be negative (0 turns it off).")) + return + } + s.mu.Lock() + defer s.mu.Unlock() + o := s.findOrg(sc.org) + if o == nil { + fail(w, 404, errors.New("organization settings only")) + return + } + o.TrafficSpikeFactor, o.TrafficDailyMB = in.SpikeFactor, in.DailyMB + s.save() + ok(w) +} diff --git a/traffic_test.go b/traffic_test.go new file mode 100644 index 0000000..306fed0 --- /dev/null +++ b/traffic_test.go @@ -0,0 +1,159 @@ +package hammerhead + +import ( + "strings" + "testing" + "time" + + "git.veillax.com/Veillax/reef/model" +) + +func TestSpiking(t *testing.T) { + const mb = 1 << 20 + cases := []struct { + name string + recent int64 + base float64 + history int64 + factor int + want bool + }{ + {"ten times usual", 200 * mb, 10 * mb, 180, 5, true}, + {"just under the factor", 49 * mb, 10 * mb, 180, 5, false}, + {"off", 500 * mb, 10 * mb, 180, 0, false}, + {"too little history", 500 * mb, 10 * mb, 30, 5, false}, + {"tiny in absolute terms", 5 * mb, 0.01 * mb, 180, 5, false}, + {"from nothing to a lot", 50 * mb, 0, 180, 5, true}, + } + for _, c := range cases { + if got := spiking(c.recent, c.base, c.history, c.factor); got != c.want { + t.Errorf("%s: spiking = %v, want %v", c.name, got, c.want) + } + } +} + +func TestTrafficStorage(t *testing.T) { + s := newFx(t).s + now := time.Now() + st := func(in, out int64, src ...model.SourceStat) model.Stat { + return model.Stat{Conns: 1, BytesIn: in, BytesOut: out, Sources: src} + } + s.db.addTrafficAt("m0", st(100, 900, model.SourceStat{IP: "203.0.113.1", Conns: 1, BytesIn: 100, BytesOut: 900}), now) + s.db.addTrafficAt("m0", st(10, 10, model.SourceStat{IP: "203.0.113.1", Conns: 1, BytesIn: 10, BytesOut: 10}, model.SourceStat{IP: "203.0.113.2", Conns: 1, BytesIn: 1}), now) + s.db.addTrafficAt("m0", st(1, 1), now.Add(-3*time.Minute)) + s.db.addTrafficAt("m1", st(7, 7), now) + + pts := s.db.trafficSeries([]string{"m0"}, "minute", 60, now) + if len(pts) != 60 || pts[59].BytesIn != 110 || pts[59].BytesOut != 910 || pts[59].Conns != 2 || pts[56].BytesIn != 1 { + t.Fatalf("minute series: last=%+v 3 back=%+v", pts[59], pts[56]) + } + if pts[59].Time != now.Unix()/60*60 || pts[58].Time != pts[59].Time-60 { + t.Errorf("bucket times: %d %d", pts[58].Time, pts[59].Time) + } + all := s.db.trafficSeries([]string{"m0", "m1"}, "hour", 24, now) + if all[23].BytesIn+all[22].BytesIn < 118 || len(all) != 24 { + t.Errorf("hour series over two ports: %+v", all[22:]) + } + if got := s.db.trafficSeries(nil, "hour", 3, now); len(got) != 3 || got[2].BytesIn != 0 { + t.Errorf("no ports: %+v", got) + } + + top := s.db.topSources([]string{"m0"}, 24, 10, now) + if len(top) != 2 || top[0].IP != "203.0.113.1" || top[0].BytesOut != 910 || top[0].Conns != 2 { + t.Fatalf("top sources: %+v", top) + } + if got := s.db.topSources([]string{"m0"}, 24, 1, now); len(got) != 1 { + t.Errorf("limit: %+v", got) + } + if got := s.db.topSources([]string{"m1"}, 24, 10, now); len(got) != 0 { + t.Errorf("another port's sources leaked: %+v", got) + } +} + +func TestTrafficRetention(t *testing.T) { + s := newFx(t).s + now := time.Now() + old := model.Stat{Conns: 1, BytesIn: 5, Sources: []model.SourceStat{{IP: "203.0.113.9", Conns: 1, BytesIn: 5}}} + s.db.addTrafficAt("m0", old, now.Add(-trafficMinutelyKeep-time.Hour)) // too old for minutes, fine for hours and sources + s.db.addTrafficAt("m0", old, now.Add(-trafficSourcesKeep-time.Hour)) // too old for sources too + s.db.addTrafficAt("m0", old, now.Add(-trafficHourlyKeep-time.Hour)) // gone everywhere + s.db.addTrafficAt("m0", old, now) + s.db.purge(30, 365) + count := func(table string) (n int) { + s.db.sql.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n) + return + } + if n := count("traffic_minutely"); n != 1 { + t.Errorf("minutely rows after purge: %d, want 1", n) + } + if n := count("traffic_sources"); n != 2 { + t.Errorf("source rows after purge: %d, want 2", n) + } + if n := count("traffic_hourly"); n != 3 { + t.Errorf("hourly rows after purge: %d, want 3", n) + } +} + +func TestRecentTraffic(t *testing.T) { + s := newFx(t).s + now := time.Now() + const mb = 1 << 20 + for h := 3 * 60; h > spikeWindow; h-- { // three hours of 1 MB a minute + s.db.addTrafficAt("m0", model.Stat{BytesIn: mb}, now.Add(-time.Duration(h)*time.Minute)) + } + s.db.addTrafficAt("m0", model.Stat{BytesIn: 100 * mb}, now) + recent, base, hist := s.db.recentTraffic("m0", now) + if recent != 100*mb || hist < 170 || base < 4.9*mb || base > 5.1*mb { // 1 MB a minute is 5 MB per window + t.Fatalf("recent=%d base=%.0f history=%d", recent, base, hist) + } + if !spiking(recent, base, hist, 5) { + t.Error("100 MB over a 5 MB norm should spike") + } + if r, b, h := s.db.recentTraffic("nothing", now); r != 0 || b != 0 || h != 0 { + t.Errorf("unknown port: %d %f %d", r, b, h) + } +} + +func TestTrafficAlertsFireAndClear(t *testing.T) { + f := newFx(t) + s := f.s + const mb = 1 << 20 + now := time.Now() + s.mu.Lock() + m := s.st.Mappings[0] + o := s.findOrg(m.OrgID) + o.TrafficSpikeFactor, o.TrafficDailyMB = 4, 500 + id := m.ID + s.mu.Unlock() + for min := 3 * 60; min > spikeWindow; min-- { + s.db.addTrafficAt(id, model.Stat{BytesIn: mb}, now.Add(-time.Duration(min)*time.Minute)) + } + + problems := func() map[string]problem { + s.mu.Lock() + defer s.mu.Unlock() + return s.problemsLocked(now) + } + if p := problems(); p["spike:"+id].msg != "" || p["quota:"+id].msg != "" { + t.Fatalf("steady traffic raised a problem: %+v", p) + } + s.db.addTrafficAt(id, model.Stat{BytesOut: 300 * mb}, now) + p := problems() + if !strings.Contains(p["spike:"+id].msg, "usual") || p["spike:"+id].kind != "traffic" { + t.Fatalf("no spike problem: %+v", p) + } + if p["quota:"+id].msg != "" { + t.Errorf("under the daily amount, yet: %+v", p["quota:"+id]) + } + s.db.addTrafficAt(id, model.Stat{BytesOut: 300 * mb}, now) + if q := problems()["quota:"+id]; !strings.Contains(q.msg, "over the") { + t.Fatalf("no quota problem: %+v", q) + } + + s.mu.Lock() + o.TrafficSpikeFactor, o.TrafficDailyMB = 0, 0 + s.mu.Unlock() + if p := problems(); p["spike:"+id].msg != "" || p["quota:"+id].msg != "" { + t.Errorf("alerts off, yet: %+v", p) + } +} diff --git a/web/index.html b/web/index.html index 8b70d41..3db60a6 100644 --- a/web/index.html +++ b/web/index.html @@ -634,7 +634,7 @@ views.ports = async () => { S.headerAction = ['Add resource', () => portForm()]; const q = h('input', {type:'search', placeholder:'Search resources…', oninput:() => { const t = q.value.toLowerCase(); table.querySelectorAll('tbody tr').forEach(tr => { tr.style.display = tr.textContent.toLowerCase().includes(t) ? '' : 'none'; }); }}); return h('div', {}, unsetDash ? h('div', {class:'notice'}, 'Some ports have a login in front of them, but the dashboard URL isn\'t set. Visitors are sent back to this address\'s login page, so set it under Server → Settings if people reach Hammerhead by a different address.') : null, - tagBar('ports', tags), rows.length > 5 ? h('div', {class:'toolbar'}, q) : null, table); + tagBar('ports', tags), rows.length ? h('div', {class:'toolbar'}, h('button', {class:'ghost', onclick:() => trafficDialog('all', 'All resources')}, 'Traffic overview')) : null, rows.length > 5 ? h('div', {class:'toolbar'}, q) : null, table); }; function spark(series) { @@ -648,6 +648,57 @@ function spark(series) { return h('div', {title:'traffic per hour, last 24 hours'}, svg); } +// Traffic chart for one resource (or "all"): stacked bars per time bucket. +function trafficChart(d) { + const ns = 'http://www.w3.org/2000/svg', W = 640, H = 170, L = 52, B = 22, T = 8, n = d.points.length; + const mk = (tag, attrs, text) => { const e = document.createElementNS(ns, tag); for (const k in attrs) e.setAttribute(k, attrs[k]); if (text != null) e.textContent = text; return e; }; + const per = d.bucketSeconds >= 3600 ? 'hour' : 'minute'; + const svg = mk('svg', {viewBox:`0 0 ${W} ${H}`, width:'100%', role:'img', 'aria-label':`Traffic per ${per}: ${bytes(d.total.bytesOut)} out and ${bytes(d.total.bytesIn)} in`}); + const max = Math.max(d.peakBucket, 1), plotH = H - B - T, bw = (W - L) / n; + for (const f of [0, .5, 1]) { + const y = T + plotH * (1 - f); + svg.append(mk('line', {x1:L, x2:W, y1:y, y2:y, stroke:'var(--border)', 'stroke-width':1}), mk('text', {x:L - 6, y:y + 4, 'text-anchor':'end', fill:'var(--text-faint)', 'font-size':11}, bytes(max * f))); + } + d.points.forEach((p, i) => { + const hOut = p.bytesOut / max * plotH, hIn = p.bytesIn / max * plotH, x = L + i * bw, g = mk('g', {}); + g.append(mk('title', {}, `${new Date(p.t * 1000).toLocaleString(undefined, {month:'short', day:'numeric', hour:'numeric', minute:per === 'minute' ? '2-digit' : undefined})}: ${bytes(p.bytesOut)} out, ${bytes(p.bytesIn)} in, ${p.conns} conn`), + mk('rect', {x:x + .5, y:T + plotH - hOut, width:Math.max(bw - 1, .5), height:hOut, fill:'var(--accent)'}), + mk('rect', {x:x + .5, y:T + plotH - hOut - hIn, width:Math.max(bw - 1, .5), height:hIn, fill:'var(--text-faint)'})); + svg.append(g); + }); + for (const i of [0, Math.floor(n / 2), n - 1]) { + const t = new Date(d.points[i].t * 1000); + svg.append(mk('text', {x:i === n - 1 ? W : L + i * bw, y:H - 6, 'text-anchor':i === 0 ? 'start' : i === n - 1 ? 'end' : 'middle', fill:'var(--text-faint)', 'font-size':11}, + per === 'minute' || n <= 24 ? t.toLocaleTimeString(undefined, {hour:'numeric', minute:'2-digit'}) : t.toLocaleDateString(undefined, {month:'short', day:'numeric'}))); + } + return svg; +} + +// A dialog with the traffic chart, totals and busiest visitors of one resource ("all" = every resource). +function trafficDialog(id, name) { + const body = h('div', {class:'stack'}, h('p', {class:'dim'}, 'Loading…')); + const close = modal('Traffic: ' + name, body, {wide:true}); + const load = async r => { + const d = await api('GET', O(`/traffic/${id}?range=${r}`)); + const per = d.bucketSeconds >= 3600 ? 'hour' : 'minute'; + const ranges = [['1h', 'Hour'], ['6h', '6 hours'], ['24h', 'Day'], ['7d', 'Week'], ['30d', 'Month']]; + const covered = d.sourcesHours >= 48 ? Math.round(d.sourcesHours / 24) + ' days' : d.sourcesHours === 1 ? 'hour' : d.sourcesHours + ' hours'; + body.replaceChildren( + h('div', {class:'tabs2'}, ranges.map(([k, l]) => h('button', {class:k === r ? 'on' : '', onclick:() => load(k).catch(e => toast(e.message))}, l))), + d.total.bytesIn + d.total.bytesOut ? h('div', {}, trafficChart(d), + h('div', {class:'small dim'}, h('span', {style:'color:var(--accent)'}, '■ '), 'To visitors (out) ', h('span', {style:'margin-left:12px;color:var(--text-faint)'}, '■ '), 'From visitors (in) · per ' + per)) : + emptyState('ports', 'No traffic in this period', 'Traffic appears here a few seconds after it happens.'), + h('div', {class:'small'}, `${d.total.conns} connections or requests, ${bytes(d.total.bytesOut)} out, ${bytes(d.total.bytesIn)} in.`), + h('h3', {}, 'Busiest visitors'), + d.sources.length ? h('div', {class:'scroll'}, h('table', {}, + h('thead', {}, h('tr', {}, ['Address', 'Connections', 'Out', 'In'].map(x => h('th', {}, x)))), + h('tbody', {}, d.sources.map(s => h('tr', {}, h('td', {class:'mono'}, s.ip), h('td', {}, s.conns), h('td', {}, bytes(s.bytesOut)), h('td', {}, bytes(s.bytesIn))))))) : h('p', {class:'dim small'}, 'None named yet.'), + h('p', {class:'dim small'}, `Counted over the last ${covered} (visitor addresses are kept for 7 days). Each report names the busiest ten per resource, so very small visitors may be missing.`), + h('div', {class:'actions form-actions'}, h('button', {class:'btn', onclick:() => close()}, 'Close'))); + }; + load('24h').catch(e => { close(); toast(e.message); }); +} + const portText = m => m.portEnd > m.port ? m.port + '-' + m.portEnd : String(m.port); function mappingRow(m, st, series) { @@ -678,6 +729,7 @@ function mappingRow(m, st, series) { })), h('td', {class:'small dim'}, m.total && m.total.conns ? [`${m.total.conns} conn`, h('br'), `${bytes(m.total.bytesIn)} in, ${bytes(m.total.bytesOut)} out`] : 'None yet', spark(series)), h('td', {}, h('div', {class:'actions'}, + m.redirectTo ? null : h('button', {class:'ghost', onclick:() => trafficDialog(m.id, m.name)}, 'Traffic'), h('button', {class:'ghost', onclick:() => m.redirectTo ? redirectForm(m) : portForm(m)}, 'Edit'), h('button', {class:'ghost', onclick: act(async () => { await api('PUT', O('/mappings/' + m.id), {...m, enabled:!m.enabled}); refresh(); })}, m.enabled ? 'Pause' : 'Enable'), web && prot ? h('button', {class:'ghost', onclick:() => shareDialog(m)}, 'Share') : null, @@ -1169,6 +1221,14 @@ function channelForm(base, existing, trusted) { h('div', {class:'actions form-actions'}, h('button', {class:'ghost', onclick:() => close()}, 'Cancel'), h('button', {class:'btn', onclick:save}, 'Save'))), {sticky:true}); } +function trafficAlertCard(base, settings) { + const spike = h('input', {type:'number', min:0, max:1000, value:settings.spikeFactor || 0}), daily = h('input', {type:'number', min:0, value:settings.dailyMb || 0}); + return card('Traffic', h('p', {class:'dim'}, 'Be told when a port suddenly carries far more than it usually does, or more than you expect in a day. These apply to every port in this organization. Leave them at 0 to switch them off.'), + h('div', {class:'grid'}, field('Spike: times the usual traffic', spike, 'Judged on the last 5 minutes against the previous hours, once a port has an hour of history and carried over 10 MB in those minutes. Try 5.'), + field('Most traffic a port may carry in a day (MB)', daily, 'Counted per port from midnight UTC, in and out together.'), + h('button', {class:'ghost', onclick: act(async () => { await api('PUT', base + '/traffic', {spikeFactor:+spike.value, dailyMb:+daily.value}); toast('Traffic alerts saved'); })}, 'Save traffic alerts'))); +} + async function alertsView(base, scoped, trusted) { const d = await api('GET', base); const channels = h('table', {}, h('thead', {}, h('tr', {}, ['Name', 'Type', 'Sends to', ''].map(x => h('th', {}, x)))), @@ -1189,6 +1249,7 @@ async function alertsView(base, scoped, trusted) { scoped ? card('Timing', h('p', {class:'dim'}, 'A problem has to last this long before you are told, so a restart or a brief blip stays quiet. You get one message when it starts and one when it clears.'), h('div', {class:'grid'}, field('Wait this many minutes first', debounce), field('Remind me every (hours)', repeat, 'Use 0 to send no reminders'), h('button', {class:'ghost', onclick: act(async () => { await api('PUT', base + '/settings', {debounceMin:+debounce.value, repeatHours:+repeat.value}); toast('Alert timing saved'); })}, 'Save timing'))) : null, + scoped ? trafficAlertCard(base, d.settings) : null, card('History', d.history.length ? h('table', {}, h('tbody', {}, d.history.map(e => h('tr', {}, h('td', {class:'dim small'}, fmtTime(e.time)), h('td', {}, h('span', {class:'tag'}, e.status)), h('td', {}, e.message))))) : emptyState('alerts', 'No alerts yet', 'Alerts that fired and recovered are listed here.'))); }