Hammerhead/traffic.go

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)
}